Let a story go two ways at once

Attempts pile up at the tip as leaves and cost nothing. The moment the
player takes the story down one the line has already moved past, the two
futures have to coexist — so `tree.fork` gives that attempt a branch of
its own, forked at the depth just before it, and the line it leaves
keeps every turn it has.

One row is inserted and one row is moved. Nothing is copied: everything
before the fork is borrowed through the lineage cached on the branch
row. Measured on a 40-turn story forked twenty times — 21 branches, 140
rows, an 80-action story — the page load costs 31,652 B against the
31,433 B the same story flat costs, and a branch is 103 B of ancestry.

Nothing derived moves either, and that is the part worth keeping: a
memory hangs off the coordinate the parent's attempt still occupies, and
the lineage caps the parent one depth short of it. The fork simply
cannot see it, so it resummarizes that ground from the text it actually
tells, without a line of bookkeeping.

Two things had to change underneath. A new node's depth now comes from
the tip of its branch rather than from the adventure-wide `index`, which
would have left a hole in a fork's path the width of the other branch.
And undo stops at the fork — the turns before it belong to the branch
this one grew out of.

`GET /branches`, `POST /branches/{id}/switch` and
`POST /actions/{id}/fork` are the endpoints SP7's tree view is drawn on.

365 tests green, 18 of them new in test_branch_forking.py.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_017Dvvqn9ZDR4ixeFPHNbww7

backend/app/routers/adventures.py

@@ -5,7 +5,7 @@ import threading
 from fastapi import APIRouter, Body, Depends, HTTPException, Request
 from fastapi.responses import StreamingResponse
 from sqlalchemy import func
-from sqlalchemy.orm import Session, load_only
+from sqlalchemy.orm import Session, load_only, undefer
 from sqlalchemy.orm.attributes import set_committed_value
 
 from .. import (
@@ -585,6 +585,18 @@ def next_index(adventure: models.Adventure) -> int:
     return context_history.max_action_index(adventure) + 1
 
 
+def next_depth(adventure: models.Adventure) -> int:
+    """Where the next node played onto this story goes: one past the tip.
+
+    Not `next_index`, which the two agreed on until SP5. `index` has to stay
+    unique across the whole adventure — it is the v1 bundle's key — so on a
+    story forked at depth 6 after twenty turns it would hand the next node
+    depth 21 and leave a fourteen-deep hole in the middle of a path. A depth is
+    a position along *this* story, and the branch is what makes it unambiguous.
+    """
+    return adventure.head_depth + 1
+
+
 def last_action(adventure: models.Adventure, db: Session) -> models.Action | None:
     """The newest action of any kind on the story being played, or None.
 
@@ -740,10 +752,7 @@ async def _generate_turn(
     # second take on turn 12 is still turn 12. (It was `retry_of.index` until
     # SP4, which held the same number; depth is the one that stays true once a
     # branch has its own numbering.)
-    # `next_index` because `tree.place_action` still derives a new node's depth
+    ai_depth = retry_of.depth if retry_of is not None else next_depth(adventure)
-    # from its legacy index while the two columns coexist; they hold the same
-    # number, and SP8 removes the question.
-    ai_depth = retry_of.depth if retry_of is not None else next_index(adventure)
     stat_schema = adventure.scenario.stat_schema if adventure.scenario else None
     if worldstate.has_schema(stat_schema):
         text, delta = worldstate.extract_delta(text)
@@ -765,7 +774,8 @@ async def _generate_turn(
         # shares its depth: it is the same turn. Two rows then hold one index,
         # which `max_action_index` (a maximum, not a count) survives, and
         # nothing else still reads the column.
-        index=retry_of.index if retry_of is not None else ai_depth,
+        index=retry_of.index if retry_of is not None else next_index(adventure),
+        depth=ai_depth,
         type="ai",
         text=text,
         reasoning=reasoning,
@@ -832,6 +842,7 @@ async def run_player_turn(
         player_action = models.Action(
             adventure_id=adventure.id,
             index=next_index(adventure),
+            depth=next_depth(adventure),
             type=payload.type,
             text=modified,
         )
@@ -1031,6 +1042,190 @@ def delete_turn(
         db.delete(attempt)
 
 
+# ---------- Branches (Phase 14, SP5) ----------
+#
+# Attempts pile up at the tip as siblings and cost nothing. One becomes a
+# *branch* at the moment the player takes the story down it and leaves the line
+# that moved past it — which is the same event as "a turn is played past it",
+# seen from the side that has to do the work. Doing it here rather than on the
+# next turn means a branch is only ever created for a divergence somebody
+# actually built on, and the line being left is never disturbed.
+
+
+def current_window(db: Session, adventure: models.Adventure) -> schemas.ActionPage:
+    actions, total, has_more = action_window(db, adventure)
+    return schemas.ActionPage(
+        actions=[schemas.ActionOut.model_validate(a) for a in actions],
+        total=total,
+        has_more=has_more,
+    )
+
+
+@router.get("/{adventure_id}/branches", response_model=list[schemas.BranchOut])
+def list_branches(
+    adventure_id: int, db: Session = Depends(get_db), user: models.User = CurrentUser
+):
+    """Every branch of the adventure, with where each one leaves its parent.
+
+    The shape a tree view is drawn from: `fork_depth` says where the line
+    splits off and `depth` where it currently ends, so the whole picture is one
+    query over `branches` plus one grouped query over `actions` — never one per
+    branch, which is how a spatial view of a hundred forks stops being a
+    hundred round trips.
+    """
+    adventure = get_adventure_or_404(adventure_id, db, user)
+    branches = (
+        db.query(models.Branch)
+        .filter(models.Branch.adventure_id == adventure.id)
+        .order_by(models.Branch.id)
+        .all()
+    )
+    owned = {
+        branch_id: (count, tip)
+        for branch_id, count, tip in db.query(
+            models.Action.branch_id,
+            func.count(models.Action.id),
+            func.max(models.Action.depth),
+        )
+        .filter(
+            models.Action.adventure_id == adventure.id,
+            models.Action.live.is_(True),
+        )
+        .group_by(models.Action.branch_id)
+        .all()
+    }
+    out = []
+    for branch in branches:
+        count, tip = owned.get(branch.id, (0, None))
+        out.append(schemas.BranchOut(
+            id=branch.id,
+            parent_branch_id=branch.parent_branch_id,
+            fork_depth=branch.fork_depth,
+            # A branch with nothing of its own sits at its fork point: that is
+            # the last node its story contains, borrowed but the tip all the
+            # same. Mirrors tree.refresh_head.
+            depth=tip if tip is not None else (
+                branch.fork_depth if branch.fork_depth is not None else tree.NO_DEPTH
+            ),
+            own_actions=count,
+            is_head=(branch.id == adventure.head_branch_id),
+            created_at=branch.created_at,
+        ))
+    return out
+
+
+@router.post(
+    "/{adventure_id}/branches/{branch_id}/switch", response_model=schemas.ActionPage
+)
+def switch_branch(
+    adventure_id: int,
+    branch_id: int,
+    db: Session = Depends(get_db),
+    user: models.User = CurrentUser,
+):
+    """Read and play a different branch of the story.
+
+    Nothing is copied and nothing is rewritten — the head pointer moves, and
+    the shared script/world state comes back to what that branch's tip left
+    behind. That last part is why a switch is safe at all: the scoreboard and
+    the RPG layer are per-adventure, so a branch that did not restore them
+    would be told a story with another branch's numbers under it (the
+    world-state cooldown clock included, which lives inside the snapshot).
+    """
+    adventure = get_adventure_or_404(adventure_id, db, user)
+    branch = db.get(models.Branch, branch_id)
+    if branch is None or branch.adventure_id != adventure.id:
+        raise HTTPException(404, "Branch not found")
+    acquire_turn_lock(adventure_id)
+    try:
+        adventure.head_branch_id = branch.id
+        tree.refresh_head(db, adventure)
+        attempts.restore_state(adventure, db_tip(db, adventure))
+        adventure.updated_at = models.utcnow()
+        db.commit()
+        db.refresh(adventure)
+        return current_window(db, adventure)
+    finally:
+        _active_turns.discard(adventure_id)
+
+
+def db_tip(db: Session, adventure: models.Adventure) -> models.Action | None:
+    """The newest node of the story as it now stands, with its outcome loaded."""
+    return (
+        db.query(models.Action)
+        .filter(
+            models.Action.adventure_id == adventure.id,
+            lineage.path_of(db, adventure).clause(models.Action),
+        )
+        .options(
+            undefer(models.Action.state_after),
+            undefer(models.Action.world_state_after),
+        )
+        .order_by(models.Action.depth.desc(), models.Action.id.desc())
+        .first()
+    )
+
+
+@router.post(
+    "/{adventure_id}/actions/{action_id}/fork", response_model=schemas.ActionPage
+)
+def fork_from_attempt(
+    adventure_id: int,
+    action_id: int,
+    db: Session = Depends(get_db),
+    user: models.User = CurrentUser,
+):
+    """Take the story down this attempt, forking a branch if it has to.
+
+    Three cases, and the first two are not forks:
+
+    * the attempt is already the one the story tells — nothing to do;
+    * its turn is the tip, so the attempts are still leaves nobody has built
+      on: switch, exactly as `/variant` does, and no branch is created;
+    * the story has moved past its turn: fork. The attempt gets a branch of its
+      own and the line it is leaving keeps every turn it has.
+    """
+    adventure = get_adventure_or_404(adventure_id, db, user)
+    action = db.get(models.Action, action_id)
+    if action is None or action.adventure_id != adventure_id:
+        raise HTTPException(404, "Action not found")
+    # Asked before the shape of the turn is, because a fork leaves the promoted
+    # attempt alone on its branch: a client that repeats the call — a double
+    # click, a retried request — must get the same answer, not a complaint that
+    # the turn it just forked has nothing to fork to.
+    if action.live and action.branch_id == adventure.head_branch_id:
+        return current_window(db, adventure)
+    if len(attempts.group(db, action)) < 2:
+        raise HTTPException(
+            400, "This turn has only one take, so there is nothing to fork to."
+        )
+    acquire_turn_lock(adventure_id)
+    try:
+        newest = last_action(adventure, db)
+        at_the_tip = (
+            newest is not None
+            and newest.branch_id == action.branch_id
+            and newest.depth == action.depth
+        )
+        if at_the_tip:
+            # The story at this coordinate is about to say something else, so
+            # what was derived from it is withdrawn — the same move retry
+            # makes. A fork needs none of that: it leaves the coordinate, and
+            # its memory, exactly where they are (see `tree.fork`).
+            memorybank.forget_node(db, adventure, action)
+            cursors.rewind_all(adventure, action.branch_id, (action.depth or 0) - 1)
+            attempts.make_live(db, adventure, action)
+        else:
+            tree.fork(db, adventure, action)
+            attempts.restore_state(adventure, action)
+        adventure.updated_at = models.utcnow()
+        db.commit()
+        db.refresh(adventure)
+        return current_window(db, adventure)
+    finally:
+        _active_turns.discard(adventure_id)
+
+
 @router.post("/{adventure_id}/undo", response_model=schemas.ActionPage)
 def undo_turn(
     adventure_id: int, db: Session = Depends(get_db), user: models.User = CurrentUser
@@ -1059,8 +1254,20 @@ def undo_turn(
             raise HTTPException(400, "Nothing to undo")
         last = newest[0]
         before_that = newest[1] if len(newest) > 1 else None
+        # Only ground this branch owns. Everything before the fork is borrowed
+        # from an ancestor and is *its* story too, so taking back a turn here
+        # must never reach across and delete a turn out of another branch. The
+        # test is on the row's own branch rather than on the fork depth,
+        # because that is the fact that decides it.
+        if last.branch_id != adventure.head_branch_id:
+            raise HTTPException(
+                400, "Nothing to undo on this branch — the turns before it "
+                     "belong to the branch it was forked from.",
+            )
         first_removed = last
-        if last.type == "ai" and before_that is not None and before_that.type in ("do", "say", "story"):
+        if (last.type == "ai" and before_that is not None
+                and before_that.type in ("do", "say", "story")
+                and before_that.branch_id == adventure.head_branch_id):
             first_removed = before_that
         # Where the story stands once the turn is gone: what the node in front
         # of the earliest removed one left behind. Read before the deletes, so
@@ -1108,6 +1315,12 @@ def export_adventure(
     # folding each group back into the `variants` array the format expects.
     # That array is the *only* remaining producer of the v1 shape: nothing in
     # the database holds one any more.
+    #
+    # A *forked* adventure has no honest v1 rendering — the format has one
+    # story and there are two — so this emits every branch's turns interleaved
+    # by index, which reads as a mangled story rather than as lost data. SP6's
+    # v2 bundle is what fixes it, and SP7 is where a player first gets a way to
+    # fork at all, so the order those two ship in is the order that matters.
     exported_actions = (
         db.query(models.Action)
         .filter(models.Action.adventure_id == adv.id)

backend/app/schemas.py

@@ -203,6 +203,26 @@ class VariantSelect(BaseModel):
     index: int = Field(ge=0)
 
 
+class BranchOut(ORMModel):
+    """One line through the story tree (Phase 14, SP5).
+
+    Enough to draw the tree and nothing more: `fork_depth` is where this line
+    leaves its parent and `depth` is where it currently ends, so a fork is two
+    numbers rather than a walk. `own_actions` counts the turns played on this
+    branch itself — the rest of its story is borrowed from its ancestors, which
+    is the whole point and also why the number is smaller than the reader
+    expects.
+    """
+
+    id: int
+    parent_branch_id: int | None = None
+    fork_depth: int | None = None
+    depth: int
+    own_actions: int = 0
+    is_head: bool = False
+    created_at: datetime
+
+
 class ActionUpdate(BaseModel):
     text: ActionText
 

backend/app/tree.py

@@ -28,6 +28,7 @@ from sqlalchemy import func, insert, update
 from sqlalchemy.orm import Session
 
 from . import models
+from .context import lineage
 
 # The head depth of an adventure with no actions. Keeps "the next node goes at
 # head_depth + 1" true with no special case, and mirrors migrations.NO_DEPTH.
@@ -89,6 +90,93 @@ def head_branch(db: Session, adventure: models.Adventure) -> models.Branch:
     return branch
 
 
+def fork(db: Session, adventure: models.Adventure, node: models.Action) -> models.Branch:
+    """Take the story down `node`, on a branch of its own.
+
+    `node` is a discarded attempt at a turn the story has already moved past.
+    Making it live where it stands would orphan every turn played after it —
+    they were written as a continuation of the attempt that won — so it moves
+    onto a new branch instead, forked from the depth just before it. The parent
+    keeps its story, complete and untouched; the new branch borrows everything
+    up to the fork and owns exactly one node.
+
+    **One row is inserted and one row is moved. Nothing is copied.** That is
+    the whole claim of the design: a fork costs a `branches` row and the
+    ancestry cached on it, whatever the story behind it is worth.
+
+    Nothing derived moves with it, and that is not an omission. A memory hangs
+    off the coordinate its block ends on, and what it describes is whatever
+    attempt was live there — which stays on the parent. From the new branch it
+    is simply out of range: the lineage caps the parent at `fork_depth`, so the
+    memory sits one depth past the border and neither the retrieval clause nor
+    the cursors can see it. The block is summarized again, from the text this
+    branch actually tells, without a line of bookkeeping.
+
+    One thing does stay behind: the attempts this node leaves. They are still
+    takes on the parent's turn, and one of them has to be the parent's story —
+    the oldest, so the line the parent keeps is the one it was written on.
+    """
+    parent = db.get(models.Branch, node.branch_id)
+    if parent is None or node.depth is None:
+        raise ValueError("cannot fork from a node that is not on a branch")
+    fork_depth = node.depth - 1
+    # The attempts this node is leaving, read *before* it moves. The session
+    # does not autoflush, so asking afterwards would still find the node here
+    # and renumber it back into the group it just left.
+    remaining = [
+        row for row in db.query(models.Action)
+        .filter(
+            models.Action.adventure_id == adventure.id,
+            models.Action.branch_id == parent.id,
+            models.Action.depth == node.depth,
+        )
+        .order_by(models.Action.variant_index, models.Action.id)
+        .all()
+        if row is not node
+    ]
+    # The parent's ancestry, every entry capped at the fork. Only the first can
+    # actually move — an older entry is already capped at the fork depth of the
+    # branch beneath it, which is shallower than any node on the parent — but
+    # capping them all says the invariant instead of relying on it.
+    inherited = [
+        [branch_id, fork_depth if cap is None else min(cap, fork_depth)]
+        for branch_id, cap in lineage.entries_of(parent)
+    ]
+    # Inserted through Core, and its lineage written second, for the reason
+    # `root_branch` spells out: this can run inside a flush, and the lineage
+    # names the row's own id.
+    new_id = db.execute(
+        insert(models.Branch).values(
+            adventure_id=adventure.id,
+            parent_branch_id=parent.id,
+            fork_depth=fork_depth,
+            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]] + inherited)
+    )
+
+    depth = node.depth
+    node.branch_id = new_id
+    node.live = True
+    node.variant_index = 0
+    node.variant_count = 0
+
+    if remaining and not any(row.live for row in remaining):
+        remaining[0].live = True
+    for i, row in enumerate(remaining):
+        row.variant_index = i
+        row.variant_count = len(remaining) if len(remaining) > 1 else 0
+
+    adventure.head_branch_id = new_id
+    adventure.head_depth = depth
+    return db.get(models.Branch, new_id)
+
+
 def place_action(
     db: Session,
     adventure: models.Adventure,

backend/tests/test_branch_clause.py

@@ -1,8 +1,11 @@
 """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
+These tests build the fork by hand: three branch rows and their nodes, written
-branch rows and their nodes, written straight to the database, arranged as the
+straight to the database, arranged as the design doc's own worked example. That
-design doc's own worked example.
+was the only way to build one when this file was written (nothing forked until
+SP5) and it stays that way now that `tree.fork` exists — a fixture that agreed
+with the code under test could not catch it being wrong. The two are checked
+against each other in `test_branch_forking.py`.
 
     branch C, tip at depth 7, lineage [(C, 7), (B, 5), (A, 3)]
     → A0 A1 A2 A3 B4 B5 C6 C7

backend/tests/test_branch_forking.py

@@ -0,0 +1,575 @@
+"""Phase 14 SP5 — continuing from a discarded attempt forks a branch.
+
+SP4 made every attempt at a turn a node. While the attempts sit at the tip they
+are leaves and cost nothing: switching between them just moves the `live` flag.
+The moment the player takes the story down one the line has already moved past,
+the two futures have to coexist — and that is a branch.
+
+What this file is really watching is the claim the whole design rests on: **a
+fork inserts one row and moves one row, whatever the story behind it is worth.**
+Everything before the fork is borrowed, not copied, and the arithmetic that
+makes borrowing readable is `lineage`.
+
+    python -m pytest tests/test_branch_forking.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 app import auth, limits, models
+from app.context import cursors, lineage
+from app.database import Base, SessionLocal, engine, get_db
+from app.main import app
+from app.providers import PromptParts
+from app.routers import adventures
+
+# `hp` moves freely; `mana` has a cooldown of 2 turns, so a clock that advances
+# when it should not shows up as a change the referee should have rejected.
+SCHEMA = {
+    "player": {
+        "hp": {"min": 0, "max": 100, "initial": 100},
+        "mana": {"min": 0, "max": 50, "initial": 50, "cooldown": 2},
+    }
+}
+
+GOLD_SCRIPT = """
+const modifier = (text) => {
+  state.gold = (state.gold || 0) + 10;
+  return { text };
+};
+modifier(text);
+"""
+
+
+class ScriptedProvider:
+    replies: list = []
+    calls = 0
+    prompts: list = []
+
+    def __init__(self, *a, **k):
+        pass
+
+    async def generate(self, parts: PromptParts, *, temperature, max_tokens):
+        index = min(ScriptedProvider.calls, len(ScriptedProvider.replies) - 1)
+        ScriptedProvider.calls += 1
+        ScriptedProvider.prompts.append((parts.system, parts.story))
+        yield ("text", ScriptedProvider.replies[index])
+
+
+@pytest.fixture()
+def client(monkeypatch):
+    Base.metadata.create_all(bind=engine)
+    setup = SessionLocal()
+    user = models.User(is_guest=False, email="fork@example.com")
+    setup.add(user)
+    setup.flush()
+    setup.add(models.Settings(user_id=user.id, api_key="enc:dummy", model="test-model"))
+    scenario = models.Scenario(user_id=user.id, title="S", stat_schema=SCHEMA)
+    setup.add(scenario)
+    setup.flush()
+    adv = models.Adventure(
+        user_id=user.id, title="Cave", scenario_id=scenario.id,
+        script_state={}, world_state={"player": {"hp": 100, "mana": 50}},
+    )
+    setup.add(adv)
+    setup.flush()
+    setup.add(models.Action(adventure_id=adv.id, index=0, type="start", text="You enter a cave."))
+    setup.add(models.AdventureScript(
+        adventure_id=adv.id, position=0, enabled=True, name="Gold", output_js=GOLD_SCRIPT,
+    ))
+    setup.commit()
+    adv_id, user_id = adv.id, user.id
+    setup.close()
+
+    ScriptedProvider.replies = ["A reply."]
+    ScriptedProvider.calls = 0
+    ScriptedProvider.prompts = []
+    monkeypatch.setattr(adventures, "OpenAICompatibleProvider", ScriptedProvider)
+    monkeypatch.setattr(auth, "resolve_provider_config", lambda s: auth.ProviderConfig(
+        "http://fake", "k", "test-model", False))
+    monkeypatch.setattr(limits, "rate_limit", lambda *a, **k: None)
+    monkeypatch.setattr(limits, "check_row_cap", lambda *a, **k: None)
+
+    def _current_user(db=Depends(get_db)):
+        return db.get(models.User, user_id)
+
+    app.dependency_overrides[auth.get_current_user] = _current_user
+    c = TestClient(app)
+    c.adv_id = adv_id
+    try:
+        yield c
+    finally:
+        app.dependency_overrides.clear()
+        adventures._active_turns.clear()
+        Base.metadata.drop_all(bind=engine)
+
+
+# ------------------------------------------------------------------ helpers
+
+def _play(client, text="look around", type="do"):
+    r = client.post(f"/api/adventures/{client.adv_id}/actions",
+                    json={"type": type, "text": text})
+    assert r.status_code == 200, r.text
+
+
+def _retry(client):
+    r = client.post(f"/api/adventures/{client.adv_id}/retry")
+    assert r.status_code == 200, r.text
+
+
+def _texts(client) -> list[str]:
+    return [a["text"] for a in client.get(f"/api/adventures/{client.adv_id}").json()["actions"]]
+
+
+def _branches(client) -> list[dict]:
+    r = client.get(f"/api/adventures/{client.adv_id}/branches")
+    assert r.status_code == 200, r.text
+    return r.json()
+
+
+def _fork(client, action_id):
+    return client.post(f"/api/adventures/{client.adv_id}/actions/{action_id}/fork")
+
+
+def _state(adv_id):
+    db = SessionLocal()
+    try:
+        adv = db.get(models.Adventure, adv_id)
+        return adv.script_state, adv.world_state
+    finally:
+        db.close()
+
+
+def _rows(adv_id) -> list[models.Action]:
+    db = SessionLocal()
+    try:
+        return (
+            db.query(models.Action)
+            .filter(models.Action.adventure_id == adv_id)
+            .order_by(models.Action.depth, models.Action.variant_index)
+            .all()
+        )
+    finally:
+        db.close()
+
+
+def _divergent_story(client):
+    """A story that retried turn 2, continued from the newer take, and left the
+    older one behind as a leaf.
+
+        start · do · [attempt one | ATTEMPT TWO] · do · next turn
+
+    Returns the id of the attempt nobody built on.
+    """
+    ScriptedProvider.replies = ["Attempt one.", "Attempt two.", "Next turn."]
+    _play(client)
+    _retry(client)
+    _play(client, "go deeper")
+    return [a.id for a in _rows(client.adv_id) if a.type == "ai" and not a.live][0]
+
+
+# ---------------------------------------------------------------- the fork
+
+def test_a_fork_inserts_one_branch_row_and_copies_no_actions(client):
+    discarded = _divergent_story(client)
+    rows_before = [a.id for a in _rows(client.adv_id)]
+    assert len(_branches(client)) == 1
+
+    r = _fork(client, discarded)
+    assert r.status_code == 200, r.text
+
+    branches = _branches(client)
+    assert len(branches) == 2, "exactly one branch row per divergence built on"
+    assert [a.id for a in _rows(client.adv_id)] == rows_before, "a fork copies nothing"
+    forked = [b for b in branches if b["parent_branch_id"] is not None][0]
+    assert forked["is_head"] is True
+    assert forked["own_actions"] == 1, "the promoted attempt, and nothing else"
+    # The fork point is the depth just before the attempt, stored rather than
+    # inferred: inferring it from where two branches first differ would be a
+    # guess, and a wrong one as soon as an attempt repeats its parent's text.
+    assert forked["fork_depth"] == forked["depth"] - 1
+
+
+def test_the_lineage_is_capped_at_the_fork_depth(client):
+    discarded = _divergent_story(client)
+    _fork(client, discarded)
+
+    db = SessionLocal()
+    try:
+        adventure = db.get(models.Adventure, client.adv_id)
+        branch = lineage.branch_of(db, adventure)
+        entries = lineage.entries_of(branch)
+        assert entries[0] == (branch.id, None), "its own nodes, to the tip"
+        assert entries[1] == (branch.parent_branch_id, branch.fork_depth)
+        assert len(entries) == 2
+    finally:
+        db.close()
+
+
+def test_both_branches_read_independently(client):
+    discarded = _divergent_story(client)
+    parent = _branches(client)[0]["id"]
+    _fork(client, discarded)
+
+    # The fork's story: everything up to the divergence, then the other take.
+    assert _texts(client) == [
+        "You enter a cave.", "> You look around.", "Attempt one.",
+    ]
+    # And the line it left is exactly as it was, turns after the fork included.
+    r = client.post(f"/api/adventures/{client.adv_id}/branches/{parent}/switch")
+    assert r.status_code == 200, r.text
+    assert _texts(client) == [
+        "You enter a cave.", "> You look around.", "Attempt two.",
+        "> You go deeper.", "Next turn.",
+    ]
+
+
+def test_the_parent_keeps_a_live_attempt_where_the_fork_left(client):
+    """Promoting the loser must not leave the parent with a hole in its story:
+    a coordinate with no live node is a turn that disappears from the read."""
+    discarded = _divergent_story(client)
+    _fork(client, discarded)
+
+    db = SessionLocal()
+    try:
+        adventure = db.get(models.Adventure, client.adv_id)
+        parent_id = db.query(models.Branch).filter_by(
+            adventure_id=adventure.id, parent_branch_id=None).one().id
+        per_coordinate = {}
+        for row in _rows(client.adv_id):
+            per_coordinate.setdefault((row.branch_id, row.depth), []).append(row)
+        for (branch_id, depth), group in per_coordinate.items():
+            live = [a for a in group if a.live]
+            assert len(live) == 1, f"branch {branch_id} depth {depth}"
+        # The parent's turn 2 is now a single take, so the pager stops offering
+        # a page through attempts that have gone their own way.
+        parent_turn = per_coordinate[(parent_id, 2)]
+        assert len(parent_turn) == 1
+        assert parent_turn[0].variant_count == 0
+    finally:
+        db.close()
+
+
+def test_playing_on_a_fork_continues_that_branchs_depths(client):
+    """A depth is a position along *this* story. Numbering the next node from
+    the adventure-wide index would leave a hole where the other branch's turns
+    are, which every windowing estimate then has to work around."""
+    discarded = _divergent_story(client)
+    _fork(client, discarded)
+    ScriptedProvider.replies = ["Onward."]
+    _play(client, "turn back")
+
+    db = SessionLocal()
+    try:
+        adventure = db.get(models.Adventure, client.adv_id)
+        path = lineage.path_of(db, adventure)
+        rows = [a for a in _rows(client.adv_id) if path.contains(a)]
+        assert [a.depth for a in sorted(rows, key=lambda a: a.depth)] == [0, 1, 2, 3, 4]
+    finally:
+        db.close()
+    assert _texts(client)[-1] == "Onward."
+
+
+# ------------------------------------------------------------- not a fork
+
+def test_forking_at_the_tip_switches_without_making_a_branch(client):
+    """Attempts nobody has built on stay leaves — that is what keeps the
+    lineage a list of divergences rather than of every retry ever."""
+    ScriptedProvider.replies = ["Attempt one.", "Attempt two."]
+    _play(client)
+    _retry(client)
+    discarded = [a.id for a in _rows(client.adv_id) if a.type == "ai" and not a.live][0]
+
+    r = _fork(client, discarded)
+    assert r.status_code == 200, r.text
+    assert len(_branches(client)) == 1, "no branch for an attempt at the tip"
+    assert _texts(client)[-1] == "Attempt one."
+
+
+def test_forking_the_attempt_already_in_the_story_does_nothing(client):
+    """Idempotent, because a client that has lost track of which take is live
+    must not be able to fork a branch per click."""
+    discarded = _divergent_story(client)
+    _fork(client, discarded)
+    promoted = [a.id for a in _rows(client.adv_id) if a.type == "ai" and a.live
+                and a.branch_id != _branches(client)[0]["id"]][0]
+    before = _texts(client)
+
+    for _ in range(3):
+        r = _fork(client, promoted)
+        assert r.status_code == 200, r.text
+    assert len(_branches(client)) == 2
+    assert _texts(client) == before
+
+
+def test_forking_a_turn_that_is_already_the_story_is_a_no_op(client):
+    ScriptedProvider.replies = ["Only take."]
+    _play(client)
+    only = [a.id for a in _rows(client.adv_id) if a.type == "ai"][0]
+
+    r = _fork(client, only)
+    assert r.status_code == 200, r.text
+    assert len(_branches(client)) == 1
+
+
+def test_forking_a_single_take_on_another_branch_is_refused(client):
+    """The only way to reach the refusal, and it names the wrong tool: a node
+    with no siblings is not a divergence, so what the caller wants is to switch
+    to the branch it is on."""
+    discarded = _divergent_story(client)
+    _fork(client, discarded)
+    parent_id = [b for b in _branches(client) if b["parent_branch_id"] is None][0]["id"]
+    stranded = [a.id for a in _rows(client.adv_id)
+                if a.branch_id == parent_id and a.depth == 2][0]
+
+    r = _fork(client, stranded)
+    assert r.status_code == 400
+    assert "one take" in r.json()["detail"]
+    assert len(_branches(client)) == 2
+
+
+# -------------------------------------------------------------- the state
+
+def test_switching_restores_the_script_and_world_state(client):
+    ScriptedProvider.replies = [
+        "A scratch.\n```state\n{\"player.hp\": -5}\n```",
+        "A beating.\n```state\n{\"player.hp\": -40}\n```",
+        "Onward.",
+    ]
+    _play(client)
+    _retry(client)
+    _play(client, "go deeper")
+    parent = _branches(client)[0]["id"]
+    on_parent = _state(client.adv_id)
+
+    discarded = [a.id for a in _rows(client.adv_id) if a.type == "ai" and not a.live][0]
+    _fork(client, discarded)
+    script_state, world_state = _state(client.adv_id)
+    assert world_state["player"]["hp"] == 95, "the attempt this branch tells"
+    assert script_state == {"gold": 10}, "one turn of gold, not three"
+
+    client.post(f"/api/adventures/{client.adv_id}/branches/{parent}/switch")
+    assert _state(client.adv_id) == on_parent
+
+
+def test_the_cooldown_clock_travels_with_the_branch(client):
+    """The world-state clock is a depth, and depths repeat across branches — so
+    it can only be right if each branch carries its own. It does, for free: the
+    clock lives inside `_meta.last_changed`, which is part of the world state a
+    switch restores."""
+    ScriptedProvider.replies = [
+        "Drained.\n```state\n{\"player.mana\": -10}\n```",
+        "Untouched.",
+        "Onward.",
+    ]
+    _play(client)
+    _retry(client)
+    _play(client, "go deeper")
+    discarded = [a.id for a in _rows(client.adv_id) if a.type == "ai" and not a.live][0]
+    on_parent = _state(client.adv_id)[1]
+    assert on_parent["_meta"]["last_changed"].get("player.mana") is None
+
+    _fork(client, discarded)
+    forked = _state(client.adv_id)[1]
+    assert forked["player"]["mana"] == 40
+    assert forked["_meta"]["last_changed"]["player.mana"] == 2
+
+    parent = [b for b in _branches(client) if b["parent_branch_id"] is None][0]["id"]
+    client.post(f"/api/adventures/{client.adv_id}/branches/{parent}/switch")
+    assert _state(client.adv_id)[1] == on_parent
+
+
+def test_a_retry_does_not_advance_the_cooldown_clock(client):
+    """SP5's one carried-over open item. A retry re-runs the *same* turn, so
+    the clock the cooldown rules read must not move — it used to be the reused
+    `index` that guaranteed this, and it is the reused depth now."""
+    ScriptedProvider.replies = [
+        "Drained.\n```state\n{\"player.mana\": -10}\n```",
+        "Drained again.\n```state\n{\"player.mana\": -10}\n```",
+    ]
+    _play(client)
+    first = _state(client.adv_id)[1]["_meta"]["last_changed"]["player.mana"]
+    _retry(client)
+    assert _state(client.adv_id)[1]["_meta"]["last_changed"]["player.mana"] == first
+    # ...and the second attempt's drain landed, rather than being rejected for
+    # a cooldown it should never have been measured against.
+    assert _state(client.adv_id)[1]["player"]["mana"] == 40
+
+
+# --------------------------------------------------------- derived work
+
+def test_a_memory_on_the_line_left_behind_is_out_of_range_on_the_fork(client):
+    """Nothing is moved or withdrawn when a branch forks. The memory hangs off
+    the coordinate the *parent's* attempt still occupies, and the lineage caps
+    the parent one depth short of it — so the fork simply cannot see it, and
+    resummarizes that ground from the text it actually tells."""
+    from app import tree
+
+    discarded = _divergent_story(client)
+    db = SessionLocal()
+    try:
+        adventure = db.get(models.Adventure, client.adv_id)
+        winner = db.query(models.Action).filter_by(
+            adventure_id=adventure.id, type="ai", live=True).order_by(
+            models.Action.depth).first()
+        memory = models.Memory(
+            adventure_id=adventure.id, text="Attempt two happened.",
+            source_start=0, source_end=winner.depth,
+        )
+        tree.attach_memory(memory, winner)
+        db.add(memory)
+        cursors.MEMORY.anchor_at(adventure, winner)
+        db.commit()
+    finally:
+        db.close()
+
+    _fork(client, discarded)
+
+    db = SessionLocal()
+    try:
+        adventure = db.get(models.Adventure, client.adv_id)
+        # Still there, untouched — it describes the parent's story, which is
+        # unchanged.
+        assert [m.text for m in db.query(models.Memory).all()] == ["Attempt two happened."]
+        path = lineage.path_of(db, adventure)
+        visible = db.query(models.Memory).filter(
+            models.Memory.adventure_id == adventure.id,
+            path.clause(models.Memory, unanchored=True),
+        ).all()
+        assert visible == [], "a sibling's memory reached this branch"
+        # ...and the mark reads as one depth short of it, so the block is due
+        # again on this branch rather than silently claimed as read.
+        assert cursors.MEMORY.depth(db, adventure) == 1
+    finally:
+        db.close()
+
+
+# ------------------------------------------------------------------- undo
+
+def test_undo_stops_at_the_fork(client):
+    """Taking back a turn on a fork must never reach across into the line it
+    was forked from — those turns are that branch's story too."""
+    discarded = _divergent_story(client)
+    _fork(client, discarded)
+    rows_before = len(_rows(client.adv_id))
+
+    r = client.post(f"/api/adventures/{client.adv_id}/undo")
+    assert r.status_code == 200, r.text
+    # The promoted attempt goes, and the player action in front of it stays:
+    # it is on the parent, and the parent still tells it.
+    assert len(_rows(client.adv_id)) == rows_before - 1
+    assert _texts(client) == ["You enter a cave.", "> You look around."]
+
+    # Nothing left of this branch's own: refuse rather than eat the parent's.
+    r = client.post(f"/api/adventures/{client.adv_id}/undo")
+    assert r.status_code == 400
+    assert "forked from" in r.json()["detail"]
+
+
+# ----------------------------------------------------------- the tree view
+
+def test_the_branch_list_is_the_tree(client):
+    discarded = _divergent_story(client)
+    _fork(client, discarded)
+    ScriptedProvider.replies = ["Onward."]
+    _play(client, "turn back")
+
+    branches = _branches(client)
+    root = [b for b in branches if b["parent_branch_id"] is None][0]
+    forked = [b for b in branches if b["parent_branch_id"] == root["id"]][0]
+    assert root["fork_depth"] is None and root["depth"] == 4
+    assert forked["fork_depth"] == 1 and forked["depth"] == 4
+    assert root["own_actions"] == 5 and forked["own_actions"] == 3
+    assert forked["is_head"] is True and root["is_head"] is False
+
+
+def test_fork_agrees_with_the_lineage_computed_by_hand(client):
+    """`test_branch_clause.make_branch` has computed a fork's lineage since SP2,
+    by hand, precisely so the fixture could not pass by agreeing with a bug in
+    the code under test. SP5 is when that code exists — so check the two
+    against each other rather than letting them drift apart."""
+    from tests.test_branch_clause import make_branch
+
+    discarded = _divergent_story(client)
+    _fork(client, discarded)
+
+    db = SessionLocal()
+    try:
+        adventure = db.get(models.Adventure, client.adv_id)
+        real = lineage.branch_of(db, adventure)
+        parent = db.get(models.Branch, real.parent_branch_id)
+        by_hand = make_branch(db, adventure, parent=parent, fork_depth=real.fork_depth)
+        # Same shape, its own id: compare the ancestry, which is the part that
+        # is arithmetic rather than allocation.
+        assert lineage.entries_of(real)[1:] == lineage.entries_of(by_hand)[1:]
+        assert lineage.entries_of(real)[0] == (real.id, None)
+    finally:
+        db.close()
+
+
+def test_a_deep_fork_chain_reads_for_what_one_branch_costs(client):
+    """Clause count is bounded by the window, not by fork count — the property
+    the whole lineage cache exists for, now measured through real forks rather
+    than hand-built rows."""
+    from tools import dbmeter
+
+    ScriptedProvider.replies = ["First take.", "Second take.", "Onward."]
+    _play(client)
+    for _ in range(8):
+        _retry(client)
+        _play(client, "onward")
+        discarded = [
+            a.id for a in _rows(client.adv_id)
+            if a.type == "ai" and not a.live
+        ]
+        if discarded:
+            assert _fork(client, discarded[-1]).status_code == 200
+            _play(client, "onward")
+    branches = _branches(client)
+    assert len(branches) > 4, "the fixture did not actually fork"
+
+    meter = dbmeter.Meter()
+    meter.attach(engine)
+    try:
+        with meter.scope("page"):
+            client.get(f"/api/adventures/{client.adv_id}")
+        page_bytes = meter.scopes[-1].total.fetched
+    finally:
+        meter.detach()
+    assert page_bytes > 0, "the meter saw nothing"
+
+    db = SessionLocal()
+    try:
+        adventure = db.get(models.Adventure, client.adv_id)
+        entries = lineage.entries_of(lineage.branch_of(db, adventure))
+        # The whole ancestry is there to be named...
+        assert len(entries) == len(branches)
+        # ...and the windowed read names as few of them as the window needs.
+        path = lineage.path_of(db, adventure)
+        assert path.prefix_covering(60) <= len(entries)
+    finally:
+        db.close()
+
+
+def test_switching_to_a_branch_of_another_adventure_is_a_404(client):
+    """A branch id names one adventure, so the two ids in the URL have to
+    agree — otherwise a guessed number reads somebody else's story."""
+    other = client.post("/api/adventures", json={"title": "Elsewhere"}).json()["id"]
+    ScriptedProvider.replies = ["Elsewhere."]
+    r = client.post(f"/api/adventures/{other}/actions", json={"type": "do", "text": "wait"})
+    assert r.status_code == 200, r.text
+    stranger = client.get(f"/api/adventures/{other}/branches").json()
+    assert len(stranger) == 1
+
+    assert client.post(
+        f"/api/adventures/{client.adv_id}/branches/{stranger[0]['id']}/switch"
+    ).status_code == 404