interactive-story/backend/tests/test_save_points.py

"""M4: Save Points, and the promise that restoring one deletes nothing.

This file is the acceptance contract for the milestone that gave a story
position a durable name. Its subject is one claim:

    A Save Point is a name for a coordinate, and restoring it is head movement.

Everything else follows. Restore keeps later history because head movement
deletes nothing (M3). Redo still walks forward afterwards because the retained
lineage is unchanged. Divergence after a restore forks through the same check
every write goes through, and the Save Point still names the same position when
it is over. Memory and context narrow and widen with the head, because the head
caps the one path all four reads share.

The tests are named for the acceptance items they discharge — D11-D14, E01-E04,
I04 and L03 in `planning/V1-ACCEPTANCE-TESTS.md`.

The world state is instrumentation here, not the subject, exactly as it is in
`test_head_cursor.py`: each scripted reply banks ten gold so that "the state at
this position" is a number a test can assert rather than a paragraph it has to
interpret. M5 replaces that machinery with genre-neutral narrative state, and
these tests then need the instrumentation moved, not the assertions removed.

    python -m pytest tests/test_save_points.py -v
"""
import pytest
from fastapi import Depends
from fastapi.testclient import TestClient

from app import limits, models
from app.context import lineage
from app.database import Base, SessionLocal, engine, get_db
from app.main import app
from app import auth
from app.routers import adventures

from fakes import GOLD_PER_TURN, GOLD_SCHEMA, ScriptedProvider, gold_replies


@pytest.fixture()
def client(monkeypatch):
    Base.metadata.create_all(bind=engine)
    setup = SessionLocal()
    user = models.User(is_guest=False, email="save@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=GOLD_SCHEMA)
    setup.add(scenario)
    setup.flush()
    adv = models.Adventure(
        user_id=user.id, title="Abbey", scenario_id=scenario.id,
        world_state={"player": {"hp": 100, "gold": 0}},
    )
    setup.add(adv)
    setup.flush()
    setup.add(models.Action(adventure_id=adv.id, type="start", text="The road forks."))
    setup.commit()
    adv_id, user_id = adv.id, user.id
    setup.close()

    ScriptedProvider.replies = gold_replies()
    monkeypatch.setattr(adventures.turns, "OpenAICompatibleProvider", ScriptedProvider)
    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.turns._active_turns.clear()
        Base.metadata.drop_all(bind=engine)


# ------------------------------------------------------------------ helpers

def _play(client, text="look around", type="do", adv_id=None):
    r = client.post(
        f"/api/adventures/{adv_id or client.adv_id}/actions",
        json={"type": type, "text": text},
    )
    assert r.status_code == 200, r.text
    return r


def _turns(client, count):
    for n in range(count):
        _play(client, f"turn {n}")


def _undo(client):
    return client.post(f"/api/adventures/{client.adv_id}/undo")


def _redo(client):
    return client.post(f"/api/adventures/{client.adv_id}/redo")


def _save(client, name="Before entering the abbey", note=None, adv_id=None):
    """Creates a Save Point at wherever the story is being read."""
    payload = {"name": name}
    if note is not None:
        payload["note"] = note
    r = client.post(
        f"/api/adventures/{adv_id or client.adv_id}/checkpoints", json=payload
    )
    assert r.status_code == 201, r.text
    return r.json()


def _list(client, adv_id=None):
    r = client.get(f"/api/adventures/{adv_id or client.adv_id}/checkpoints")
    assert r.status_code == 200, r.text
    return r.json()


def _restore(client, checkpoint_id, adv_id=None):
    return client.post(
        f"/api/adventures/{adv_id or client.adv_id}/checkpoints/{checkpoint_id}/restore"
    )


def _adventure(client, adv_id=None) -> dict:
    r = client.get(f"/api/adventures/{adv_id or client.adv_id}")
    assert r.status_code == 200, r.text
    return r.json()


def _texts(client, adv_id=None) -> list[str]:
    return [a["text"] for a in _adventure(client, adv_id)["actions"]]


def _rows(adv_id) -> list[models.Action]:
    """Every action row, story or not, live or not, head or no head."""
    db = SessionLocal()
    try:
        return (
            db.query(models.Action)
            .filter(models.Action.adventure_id == adv_id)
            .order_by(models.Action.branch_id, models.Action.depth, models.Action.id)
            .all()
        )
    finally:
        db.close()


def _gold(adv_id) -> int:
    db = SessionLocal()
    try:
        adv = db.get(models.Adventure, adv_id)
        return (adv.world_state or {}).get("player", {}).get("gold", 0)
    finally:
        db.close()


def _head(adv_id) -> tuple[int, int]:
    db = SessionLocal()
    try:
        adv = db.get(models.Adventure, adv_id)
        return adv.head_branch_id, adv.head_depth
    finally:
        db.close()


def _restart(client):
    """Simulates stopping and restarting the application.

    A Save Point is a row, so what a restart has to prove is that nothing about
    it lived in the process. Every session this file opens is closed again, so
    dropping the client's own session and reading through a new one is what a
    restart changes: no cached adventure, no cached head, no in-memory list.
    """
    client.close()
    adventures.turns._active_turns.clear()
    fresh = TestClient(app)
    fresh.adv_id = client.adv_id
    return fresh


def _export(client, adv_id=None) -> dict:
    r = client.get(f"/api/adventures/{adv_id or client.adv_id}/export")
    assert r.status_code == 200, r.text
    return r.json()


def _import(client, bundle) -> dict:
    r = client.post("/api/adventures/import", json=bundle)
    assert r.status_code == 201, r.text
    return r.json()


def _story_of(adv_id) -> list[str]:
    """The story an adventure tells, read through its own head."""
    db = SessionLocal()
    try:
        adventure = db.get(models.Adventure, adv_id)
        rows = (
            db.query(models.Action)
            .filter(
                models.Action.adventure_id == adv_id,
                lineage.path_of(db, adventure).clause(models.Action),
            )
            .order_by(models.Action.depth, models.Action.id)
            .all()
        )
        return [a.text for a in rows]
    finally:
        db.close()


# --------------------------------------------------------- D11: named, durable

def test_d11_a_named_save_point_survives_a_restart(client):
    """The acceptance test names the Save Point, so this one does too."""
    _turns(client, 3)
    made = _save(client, "Before entering the abbey")
    assert made["name"] == "Before entering the abbey"

    after = _restart(client)
    try:
        kept = _list(after)
        assert [c["name"] for c in kept] == ["Before entering the abbey"]
        assert kept[0]["id"] == made["id"]
        assert (kept[0]["branch_id"], kept[0]["depth"]) == (
            made["branch_id"], made["depth"]
        )
    finally:
        after.close()


def test_a_save_point_records_the_position_the_story_is_read_at(client):
    """The active head, not the retained tip. A Save Point made after an Undo
    names the undone position, because that is the turn the reader is looking
    at — a distinction that only exists because M3 stopped Undo deleting."""
    _turns(client, 5)
    _undo(client)
    _undo(client)
    branch, depth = _head(client.adv_id)

    made = _save(client, "Two turns back")

    assert (made["branch_id"], made["depth"]) == (branch, depth)
    assert made["turn"] == depth + 1


def test_a_save_point_needs_a_name(client):
    _turns(client, 1)
    for blank in ("", "   ", "\n\t "):
        r = client.post(
            f"/api/adventures/{client.adv_id}/checkpoints", json={"name": blank}
        )
        assert r.status_code == 400, r.text
        assert "needs a name" in r.json()["detail"]
    assert _list(client) == []


def test_a_name_is_stored_trimmed(client):
    _turns(client, 1)
    made = _save(client, "  Before entering the abbey  ")
    assert made["name"] == "Before entering the abbey"


def test_an_empty_story_has_no_position_to_save(client):
    """The opening node is the campaign's first position, so this refuses only
    the case where the head rests in front of every turn."""
    _undo(client)  # nothing to undo; the head is already on the opening
    db = SessionLocal()
    try:
        adv = db.get(models.Adventure, client.adv_id)
        adv.head_depth = lineage.NO_DEPTH
        db.commit()
    finally:
        db.close()

    r = client.post(
        f"/api/adventures/{client.adv_id}/checkpoints", json={"name": "Nowhere"}
    )
    assert r.status_code == 400
    assert "no turn here to save" in r.json()["detail"]


# ------------------------------------------------------------ D12: restoring

def test_d12_restore_returns_the_transcript_and_the_state(client):
    _turns(client, 2)
    told = _texts(client)
    banked = _gold(client.adv_id)
    made = _save(client, "Before entering the abbey")

    _turns(client, 3)
    assert _gold(client.adv_id) == banked + 3 * GOLD_PER_TURN

    page = _restore(client, made["id"])
    assert page.status_code == 200, page.text

    assert [a["text"] for a in page.json()["actions"]] == told
    assert _texts(client) == told
    assert _gold(client.adv_id) == banked


def test_restore_is_the_same_head_movement_undo_makes(client):
    """The mechanism claim, measured rather than asserted in prose. Restoring to
    a position and undoing to it must leave the campaign in the same state, both
    halves of the head included."""
    _turns(client, 4)
    made = _save(client, "Here")
    _turns(client, 2)

    _restore(client, made["id"])
    by_restore = (_head(client.adv_id), _gold(client.adv_id), _texts(client))

    _redo(client)
    _redo(client)
    _undo(client)
    _undo(client)
    by_undo = (_head(client.adv_id), _gold(client.adv_id), _texts(client))

    assert by_restore == by_undo


def test_restoring_the_same_save_point_repeatedly_is_stable(client):
    _turns(client, 3)
    made = _save(client, "Here")
    _turns(client, 2)

    seen = []
    for _ in range(3):
        assert _restore(client, made["id"]).status_code == 200
        seen.append((_head(client.adv_id), _gold(client.adv_id), _texts(client)))
    assert seen[0] == seen[1] == seen[2]


def test_a_save_point_at_the_current_tip_restores_to_a_story_that_never_moved(client):
    _turns(client, 3)
    made = _save(client, "Right here")
    before = (_head(client.adv_id), _gold(client.adv_id), _texts(client))

    assert _restore(client, made["id"]).status_code == 200

    assert (_head(client.adv_id), _gold(client.adv_id), _texts(client)) == before


def test_a_save_point_at_the_campaign_opening_restores_to_the_opening(client):
    """The floor Undo stops at is a position like any other, and naming it must
    not need a special case."""
    _turns(client, 3)
    for _ in range(3):
        _undo(client)
    assert _texts(client) == ["The road forks."]
    made = _save(client, "The very beginning")

    _redo(client)
    _redo(client)
    assert len(_texts(client)) == 5

    assert _restore(client, made["id"]).status_code == 200
    assert _texts(client) == ["The road forks."]
    assert _gold(client.adv_id) == 0


def test_restore_after_undo_and_redo_activity_lands_where_the_name_says(client):
    _turns(client, 5)
    _undo(client)
    _undo(client)
    made = _save(client, "Amid the undoing")
    at_save = (_head(client.adv_id), _gold(client.adv_id), _texts(client))

    _redo(client)
    _undo(client)
    _undo(client)
    _redo(client)
    _redo(client)

    assert _restore(client, made["id"]).status_code == 200
    assert (_head(client.adv_id), _gold(client.adv_id), _texts(client)) == at_save


def test_several_save_points_at_different_positions_each_restore_to_their_own(client):
    marks = []
    for n in range(4):
        _play(client, f"turn {n}")
        marks.append((_save(client, f"After turn {n}"), _texts(client), _gold(client.adv_id)))

    for mark, told, banked in reversed(marks):
        assert _restore(client, mark["id"]).status_code == 200
        assert _texts(client) == told
        assert _gold(client.adv_id) == banked


def test_two_save_points_may_name_the_same_position(client):
    """No uniqueness is imposed on names or on positions. Nothing in the product
    requirements asks for it, and two names for one turn is a reasonable thing
    for a player to want."""
    _turns(client, 2)
    first = _save(client, "Before the abbey")
    second = _save(client, "Where I keep dying")

    assert first["id"] != second["id"]
    assert (first["branch_id"], first["depth"]) == (second["branch_id"], second["depth"])

    _turns(client, 2)
    for made in (first, second):
        assert _restore(client, made["id"]).status_code == 200
        assert len(_texts(client)) == 5


# ---------------------------------------- D13: restore does not delete history

def test_d13_restore_deletes_no_accepted_history(client):
    """The measurement, on row identity rather than on a count: every row that
    existed before the restore is still there afterwards, and it is the same
    row."""
    _turns(client, 2)
    made = _save(client, "Before entering the abbey")
    _turns(client, 3)
    before = {a.id for a in _rows(client.adv_id)}

    assert _restore(client, made["id"]).status_code == 200

    after = {a.id for a in _rows(client.adv_id)}
    assert after == before
    assert len(after) == 11  # the opening, plus two rows for each of five turns


def test_d13_the_retained_continuation_can_still_be_redone(client):
    """Restore is not a decision to abandon anything, so the future it steps
    behind is still the continuation this story tells. `STORY-BRANCH-SEMANTICS`
    §20."""
    _turns(client, 2)
    made = _save(client, "Before entering the abbey")
    _turns(client, 3)
    whole = _texts(client)
    tip_gold = _gold(client.adv_id)

    _restore(client, made["id"])
    assert _adventure(client)["can_redo"] is True

    for _ in range(3):
        assert _redo(client).status_code == 200
    assert _texts(client) == whole
    assert _gold(client.adv_id) == tip_gold


def test_d13_a_different_continuation_forks_and_keeps_the_old_future(client):
    """The other half of D13. The fork happens on the first write below the
    restored head, not because Restore was clicked."""
    _turns(client, 2)
    made = _save(client, "Before entering the abbey")
    _turns(client, 3)
    old_future = {a.id for a in _rows(client.adv_id)}
    branches_before = _branch_count(client.adv_id)

    _restore(client, made["id"])
    # Restore itself created nothing.
    assert _branch_count(client.adv_id) == branches_before

    ScriptedProvider.replies = ["Through the side door.\n```state\n{\"player.gold\": 1}\n```"]
    _play(client, "go around the back")

    # The write forked...
    assert _branch_count(client.adv_id) == branches_before + 1
    # ...ordinary Redo no longer offers the displaced future...
    assert _adventure(client)["can_redo"] is False
    assert _redo(client).status_code == 400
    # ...and not one row of it was deleted to achieve that.
    assert old_future <= {a.id for a in _rows(client.adv_id)}
    assert _texts(client)[-1].startswith("Through the side door.")


def test_d13_the_save_point_still_names_the_same_position_after_divergence(client):
    """`STORY-BRANCH-SEMANTICS.md` §19: a divergence does not disturb a Save
    Point. It still points at the turn it was made on, and restoring it now
    returns to the *new* line's reading of that position."""
    _turns(client, 2)
    made = _save(client, "Before entering the abbey")
    told = _texts(client)
    _turns(client, 3)

    _restore(client, made["id"])
    ScriptedProvider.replies = ["Through the side door.\n```state\n{\"player.gold\": 1}\n```"]
    _play(client, "go around the back")

    kept = _list(client)
    assert len(kept) == 1
    assert (kept[0]["branch_id"], kept[0]["depth"]) == (made["branch_id"], made["depth"])

    assert _restore(client, made["id"]).status_code == 200
    assert _texts(client) == told
    # The new continuation is what Redo walks into now, not the displaced one.
    assert _adventure(client)["can_redo"] is True
    _redo(client)
    assert _texts(client)[-1].startswith("Through the side door.")


def test_a_save_point_on_a_line_the_story_left_still_restores(client):
    """The case that needs the head's other half to move.

    A Save Point survives divergence, so one can name a position on a future the
    story has since displaced — and no amount of depth movement reaches a branch
    the current path does not contain. Restoring it moves the line as well, the
    same single assignment a branch switch makes, and still forks nothing.
    """
    _turns(client, 2)
    fork_point = _save(client, "The fork")
    _turns(client, 3)
    deep = _save(client, "Down the old road")
    old_story = _texts(client)
    old_gold = _gold(client.adv_id)
    old_branch, old_depth = _head(client.adv_id)

    _restore(client, fork_point["id"])
    ScriptedProvider.replies = ["Through the side door.\n```state\n{\"player.gold\": 1}\n```"]
    _play(client, "go around the back")
    new_branch, _ = _head(client.adv_id)
    assert new_branch != old_branch

    # The displaced Save Point is not on the line being read...
    listed = {c["id"]: c for c in _list(client)}
    assert listed[deep["id"]]["on_path"] is False
    assert listed[fork_point["id"]]["on_path"] is True
    assert listed[deep["id"]]["resolved"] is True

    # ...and restoring it goes back to the line it names, with its own state.
    assert _restore(client, deep["id"]).status_code == 200
    assert _head(client.adv_id) == (old_branch, old_depth)
    assert _texts(client) == old_story
    assert _gold(client.adv_id) == old_gold


def test_restore_onto_an_inherited_position_keeps_the_line_being_read(client):
    """A Save Point in the shared prefix must not drag the reader back onto the
    ancestor. The turn is the same row either way; which continuation follows it
    is not, and the active line is the one the reader chose."""
    _turns(client, 4)
    _undo(client)
    _undo(client)
    ScriptedProvider.replies = ["A new road.\n```state\n{\"player.gold\": 1}\n```"]
    _play(client, "the other way")
    new_branch, _ = _head(client.adv_id)

    # Made now, on the new line, but naming a turn that physically lives on the
    # branch the story left.
    _undo(client)
    _undo(client)
    _undo(client)
    made = _save(client, "In the shared past")
    assert made["branch_id"] != new_branch

    _redo(client)
    _redo(client)
    assert _restore(client, made["id"]).status_code == 200

    # Still reading the new line, so Redo walks up the shared past and on into
    # the new continuation — depth 6 holds "A new road." on this line and the
    # displaced "Take 3." on the one the story left.
    assert _head(client.adv_id)[0] == new_branch
    for _ in range(3):
        assert _adventure(client)["can_redo"] is True
        assert _redo(client).status_code == 200
    assert _texts(client)[-1].startswith("A new road.")


def _branch_count(adv_id) -> int:
    db = SessionLocal()
    try:
        return db.query(models.Branch).filter_by(adventure_id=adv_id).count()
    finally:
        db.close()


# ------------------------------------------------------------- D14: deleting

def test_d14_delete_removes_the_pointer_and_no_story(client):
    _turns(client, 3)
    made = _save(client, "Before entering the abbey")
    _turns(client, 2)
    rows = {a.id for a in _rows(client.adv_id)}
    told = _texts(client)
    head_before = _head(client.adv_id)

    r = client.delete(f"/api/adventures/{client.adv_id}/checkpoints/{made['id']}")
    assert r.status_code == 204, r.text

    assert _list(client) == []
    assert {a.id for a in _rows(client.adv_id)} == rows
    assert _texts(client) == told
    assert _head(client.adv_id) == head_before
    assert _branch_count(client.adv_id) >= 1


def test_deleting_one_save_point_leaves_the_others(client):
    _turns(client, 1)
    first = _save(client, "One")
    _turns(client, 1)
    second = _save(client, "Two")

    client.delete(f"/api/adventures/{client.adv_id}/checkpoints/{first['id']}")

    assert [c["id"] for c in _list(client)] == [second["id"]]


def test_deleting_a_save_point_twice_is_a_404(client):
    _turns(client, 1)
    made = _save(client, "One")
    path = f"/api/adventures/{client.adv_id}/checkpoints/{made['id']}"
    assert client.delete(path).status_code == 204
    assert client.delete(path).status_code == 404


# --------------------------------------------------------------- renaming

def test_rename_changes_the_label_and_not_the_coordinate(client):
    """`STORY-BRANCH-SEMANTICS.md` §23."""
    _turns(client, 3)
    made = _save(client, "Before entering the abbey")
    _turns(client, 2)

    r = client.patch(
        f"/api/adventures/{client.adv_id}/checkpoints/{made['id']}",
        json={"name": "  Before the abbey, second try  "},
    )
    assert r.status_code == 200, r.text
    renamed = r.json()
    assert renamed["name"] == "Before the abbey, second try"
    assert (renamed["branch_id"], renamed["depth"]) == (made["branch_id"], made["depth"])

    # And it still restores to exactly the position it always did.
    _restore(client, made["id"])
    assert len(_texts(client)) == 7


def test_rename_moves_no_story_and_no_head(client):
    _turns(client, 3)
    made = _save(client, "One")
    rows = {a.id for a in _rows(client.adv_id)}
    head_before = _head(client.adv_id)

    client.patch(
        f"/api/adventures/{client.adv_id}/checkpoints/{made['id']}",
        json={"name": "Another name"},
    )

    assert {a.id for a in _rows(client.adv_id)} == rows
    assert _head(client.adv_id) == head_before


def test_rename_refuses_a_blank_name(client):
    _turns(client, 1)
    made = _save(client, "One")
    r = client.patch(
        f"/api/adventures/{client.adv_id}/checkpoints/{made['id']}", json={"name": "  "}
    )
    assert r.status_code == 400
    assert _list(client)[0]["name"] == "One"


def test_a_note_can_be_kept_and_edited(client):
    """`DATA-MODEL.md` §8's optional notes, and `BROWSER-UX-SPEC.md` §24's
    optional second field."""
    _turns(client, 1)
    made = _save(client, "One", note="the door was locked")
    assert made["note"] == "the door was locked"

    r = client.patch(
        f"/api/adventures/{client.adv_id}/checkpoints/{made['id']}",
        json={"note": "the door was barred"},
    )
    assert r.status_code == 200
    assert r.json()["note"] == "the door was barred"
    assert r.json()["name"] == "One"


# ------------------------------------------------------ ownership and errors

def test_a_save_point_cannot_be_reached_through_another_campaign(client):
    """A coordinate from another campaign names a different story's turn. The id
    is matched against the adventure in the path, so this is a 404 rather than a
    restore of the wrong story."""
    _turns(client, 3)
    mine = _save(client, "Mine")

    other = client.post(
        "/api/adventures", json={"title": "Another campaign"}
    )
    assert other.status_code in (200, 201), other.text
    other_id = other.json()["id"]

    for method, path in (
        ("post", f"/api/adventures/{other_id}/checkpoints/{mine['id']}/restore"),
        ("patch", f"/api/adventures/{other_id}/checkpoints/{mine['id']}"),
        ("delete", f"/api/adventures/{other_id}/checkpoints/{mine['id']}"),
    ):
        call = getattr(client, method)
        r = call(path, json={"name": "x"}) if method == "patch" else call(path)
        assert r.status_code == 404, (path, r.text)

    # It is untouched, and still restores in its own campaign.
    assert [c["id"] for c in _list(client)] == [mine["id"]]
    assert _restore(client, mine["id"]).status_code == 200


def test_an_unknown_save_point_is_a_404(client):
    _turns(client, 1)
    assert _restore(client, 999_999).status_code == 404
    assert client.delete(
        f"/api/adventures/{client.adv_id}/checkpoints/999999"
    ).status_code == 404


def test_a_save_point_whose_turn_is_gone_refuses_rather_than_approximating(client):
    """Moving the head to the nearest surviving turn would be worse than doing
    nothing: a Save Point that silently means somewhere else."""
    _turns(client, 3)
    made = _save(client, "Before entering the abbey")
    _turns(client, 2)
    head_before = _head(client.adv_id)

    db = SessionLocal()
    try:
        doomed = (
            db.query(models.Action)
            .filter_by(
                adventure_id=client.adv_id,
                branch_id=made["branch_id"],
                depth=made["depth"],
            )
            .all()
        )
        assert doomed
        for row in doomed:
            db.delete(row)
        db.commit()
    finally:
        db.close()

    listed = _list(client)
    assert listed[0]["resolved"] is False

    r = _restore(client, made["id"])
    assert r.status_code == 409
    assert "no longer part of this story" in r.json()["detail"]
    # And nothing moved.
    assert _head(client.adv_id) == head_before


def test_a_branch_a_save_point_names_cannot_be_deleted(client):
    """`STORY-BRANCH-SEMANTICS.md` §19: a named Save Point remains until it is
    explicitly deleted — and §28 says even a future cleanup feature must retain
    paths a checkpoint references.

    So the branch delete is refused rather than taking the Save Point with it.
    The alternative, a silent cascade, breaks §19 in the way that matters least
    visibly: the story is the thing the user asked to delete, and the named
    moments would go without ever being mentioned.
    """
    _turns(client, 2)
    _undo(client)
    _undo(client)
    ScriptedProvider.replies = ["A new road.\n```state\n{\"player.gold\": 1}\n```"]
    _play(client, "the other way")
    forked = _save(client, "On the new line")
    doomed_branch = forked["branch_id"]

    root_id = _root_branch(client.adv_id)
    assert doomed_branch != root_id
    client.post(f"/api/adventures/{client.adv_id}/branches/{root_id}/switch")

    r = client.delete(f"/api/adventures/{client.adv_id}/branches/{doomed_branch}")
    assert r.status_code == 409, r.text
    detail = r.json()["detail"]
    # The message names the Save Point, so the user does not have to hunt.
    assert "On the new line" in detail
    assert "does not delete any story" in detail

    # Nothing happened: the Save Point, the branch and the story all remain.
    assert [c["id"] for c in _list(client)] == [forked["id"]]
    assert _branch_count(client.adv_id) == 2
    assert any(a.branch_id == doomed_branch for a in _rows(client.adv_id))


def test_deleting_the_save_point_then_lets_the_branch_go(client):
    """The refusal has to be recoverable, or it is just a wall. Deleting the
    Save Point deletes no story (§25), so the cost of the recovery is a click."""
    _turns(client, 2)
    _undo(client)
    _undo(client)
    ScriptedProvider.replies = ["A new road.\n```state\n{\"player.gold\": 1}\n```"]
    _play(client, "the other way")
    forked = _save(client, "On the new line")
    doomed_branch = forked["branch_id"]
    root_id = _root_branch(client.adv_id)
    client.post(f"/api/adventures/{client.adv_id}/branches/{root_id}/switch")

    assert client.delete(
        f"/api/adventures/{client.adv_id}/branches/{doomed_branch}"
    ).status_code == 409
    rows_with_story = len(_rows(client.adv_id))

    # Delete the Save Point explicitly...
    assert client.delete(
        f"/api/adventures/{client.adv_id}/checkpoints/{forked['id']}"
    ).status_code == 204
    # ...which took no story with it...
    assert len(_rows(client.adv_id)) == rows_with_story

    # ...and now the branch can go.
    r = client.delete(f"/api/adventures/{client.adv_id}/branches/{doomed_branch}")
    assert r.status_code in (200, 204), r.text
    assert _branch_count(client.adv_id) == 1


def test_a_branch_no_save_point_names_still_deletes(client):
    """The guard must not turn into a general refusal to delete branches."""
    _turns(client, 2)
    _undo(client)
    _undo(client)
    ScriptedProvider.replies = ["A new road.\n```state\n{\"player.gold\": 1}\n```"]
    _play(client, "the other way")
    forked_branch = _head(client.adv_id)[0]
    root_id = _root_branch(client.adv_id)
    client.post(f"/api/adventures/{client.adv_id}/branches/{root_id}/switch")

    assert _list(client) == []
    r = client.delete(f"/api/adventures/{client.adv_id}/branches/{forked_branch}")
    assert r.status_code in (200, 204), r.text
    assert _branch_count(client.adv_id) == 1


def test_a_save_point_on_a_descendant_also_protects_the_branch(client):
    """Deleting a branch takes everything forked from it, so the check has to
    cover the subtree. A guard that looked only at the named branch would let a
    Save Point on a child be deleted without a word — the exact failure the
    guard exists to prevent, one level down."""
    _turns(client, 4)
    _undo(client)
    _undo(client)
    ScriptedProvider.replies = ["Second line.\n```state\n{\"player.gold\": 1}\n```"]
    _play(client, "second line")
    middle_branch = _head(client.adv_id)[0]
    _undo(client)
    ScriptedProvider.replies = ["Third line.\n```state\n{\"player.gold\": 1}\n```"]
    _play(client, "third line")
    deepest = _save(client, "Down on the deepest line")
    assert deepest["branch_id"] != middle_branch

    root_id = _root_branch(client.adv_id)
    client.post(f"/api/adventures/{client.adv_id}/branches/{root_id}/switch")

    # Deleting the *middle* branch would take the deepest one with it.
    r = client.delete(f"/api/adventures/{client.adv_id}/branches/{middle_branch}")
    assert r.status_code == 409, r.text
    assert "Down on the deepest line" in r.json()["detail"]
    assert [c["id"] for c in _list(client)] == [deepest["id"]]


def test_a_save_point_elsewhere_does_not_block_an_unrelated_branch(client):
    """The guard is scoped to the subtree being deleted, not to the campaign."""
    _turns(client, 3)
    elsewhere = _save(client, "Safe on the root")
    _undo(client)
    _undo(client)
    ScriptedProvider.replies = ["A new road.\n```state\n{\"player.gold\": 1}\n```"]
    _play(client, "the other way")
    forked_branch = _head(client.adv_id)[0]
    root_id = _root_branch(client.adv_id)
    client.post(f"/api/adventures/{client.adv_id}/branches/{root_id}/switch")

    assert elsewhere["branch_id"] == root_id
    r = client.delete(f"/api/adventures/{client.adv_id}/branches/{forked_branch}")
    assert r.status_code in (200, 204), r.text
    # The unrelated Save Point is untouched and still restores.
    assert [c["id"] for c in _list(client)] == [elsewhere["id"]]
    assert _restore(client, elsewhere["id"]).status_code == 200


def test_the_refusal_names_several_save_points_without_running_on(client):
    """A long list is truncated so the message stays a sentence someone reads."""
    _turns(client, 2)
    _undo(client)
    _undo(client)
    ScriptedProvider.replies = ["A new road.\n```state\n{\"player.gold\": 1}\n```"]
    _play(client, "the other way")
    for n in range(5):
        _save(client, f"Point {n}")
    doomed_branch = _head(client.adv_id)[0]
    root_id = _root_branch(client.adv_id)
    client.post(f"/api/adventures/{client.adv_id}/branches/{root_id}/switch")

    detail = client.delete(
        f"/api/adventures/{client.adv_id}/branches/{doomed_branch}"
    ).json()["detail"]
    assert "Point 0" in detail and "2 more" in detail
    assert "Point 4" not in detail
    assert len(_list(client)) == 5


def _root_branch(adv_id) -> int:
    db = SessionLocal()
    try:
        return (
            db.query(models.Branch)
            .filter_by(adventure_id=adv_id, parent_branch_id=None)
            .one()
            .id
        )
    finally:
        db.close()


# --------------------------------------------- E-series: lineage and memory

def test_e01_a_memory_past_a_restored_head_stops_being_retrievable(client):
    """The M3 chokepoint, exercised through the M4 door. Nothing prunes a memory
    here — it stops matching the capped path, and starts again on Redo, without
    being deleted or re-embedded."""
    _turns(client, 2)
    made = _save(client, "Before entering the abbey")
    _turns(client, 3)

    branch, depth = _head(client.adv_id)
    db = SessionLocal()
    try:
        db.add(models.Memory(
            adventure_id=client.adv_id,
            text="Mara learns the location of the key.",
            branch_id=branch, depth=depth,
        ))
        db.commit()
    finally:
        db.close()

    assert _visible_memories(client) == ["Mara learns the location of the key."]

    _restore(client, made["id"])
    assert _visible_memories(client) == []
    # Not deleted — still a row, still embedded as it was.
    assert _memory_rows(client.adv_id) == 1

    for _ in range(3):
        _redo(client)
    assert _visible_memories(client) == ["Mara learns the location of the key."]
    assert _memory_rows(client.adv_id) == 1


def test_e01_an_old_futures_memory_stays_out_of_a_new_continuation(client):
    """After restore plus a divergent write, the displaced line's memory must not
    become eligible on the line now being read."""
    _turns(client, 2)
    made = _save(client, "Before entering the abbey")
    _turns(client, 3)

    branch, depth = _head(client.adv_id)
    db = SessionLocal()
    try:
        db.add(models.Memory(
            adventure_id=client.adv_id,
            text="Mara learns the location of the key.",
            branch_id=branch, depth=depth,
        ))
        db.commit()
    finally:
        db.close()

    _restore(client, made["id"])
    ScriptedProvider.replies = ["Through the side door.\n```state\n{\"player.gold\": 1}\n```"]
    _play(client, "go around the back")

    assert _visible_memories(client) == []
    # Play on: it must not reappear as the new line grows past the old depth.
    ScriptedProvider.replies = gold_replies("New")
    _turns(client, 3)
    assert _visible_memories(client) == []
    assert _memory_rows(client.adv_id) == 1


def test_e04_the_transcript_after_a_restore_holds_only_the_active_lineage(client):
    """Everything a read can see comes through the one capped path, so the
    displaced continuation is absent from the transcript rather than filtered out
    of it."""
    _turns(client, 2)
    made = _save(client, "Before entering the abbey")
    ScriptedProvider.replies = gold_replies("Old")
    _turns(client, 3)
    displaced = [t for t in _texts(client) if t.startswith("Old")]
    assert displaced

    _restore(client, made["id"])
    ScriptedProvider.replies = ["Through the side door.\n```state\n{\"player.gold\": 1}\n```"]
    _play(client, "go around the back")
    ScriptedProvider.replies = gold_replies("New")
    _turns(client, 2)

    told = _texts(client)
    assert not any(t.startswith("Old") for t in told)
    assert any(t.startswith("New") for t in told)
    # The rows are still there; they are simply not on this path.
    assert any(a.text.startswith("Old") for a in _rows(client.adv_id))


def _visible_memories(client) -> list[str]:
    r = client.get(f"/api/adventures/{client.adv_id}/memories")
    assert r.status_code == 200, r.text
    db = SessionLocal()
    try:
        adventure = db.get(models.Adventure, client.adv_id)
        path = lineage.path_of(db, adventure)
        rows = (
            db.query(models.Memory)
            .filter(
                models.Memory.adventure_id == client.adv_id,
                path.clause(models.Memory),
            )
            .order_by(models.Memory.id)
            .all()
        )
        return [m.text for m in rows]
    finally:
        db.close()


def _memory_rows(adv_id) -> int:
    db = SessionLocal()
    try:
        return db.query(models.Memory).filter_by(adventure_id=adv_id).count()
    finally:
        db.close()


# ------------------------------------------------------ I04: export / import

def test_i04_named_save_points_survive_export_and_import(client):
    _turns(client, 4)
    _undo(client)
    _undo(client)
    early = _save(client, "Before entering the abbey", note="the door was locked")
    _redo(client)
    late = _save(client, "In the cloister")
    told = _texts(client)

    bundle = _export(client)
    assert [c["name"] for c in bundle["checkpoints"]] == [
        "Before entering the abbey", "In the cloister",
    ]

    imported = _import(client, bundle)
    copied = _list(client, adv_id=imported["id"])

    assert sorted(c["name"] for c in copied) == [
        "Before entering the abbey", "In the cloister",
    ]
    by_name = {c["name"]: c for c in copied}
    assert by_name["Before entering the abbey"]["note"] == "the door was locked"
    # The coordinates point into the imported story, not the original's rows.
    assert by_name["Before entering the abbey"]["depth"] == early["depth"]
    assert by_name["In the cloister"]["depth"] == late["depth"]
    assert all(c["resolved"] for c in copied)
    assert {c["branch_id"] for c in copied} & {
        b.id for b in _branch_rows(imported["id"])
    } == {c["branch_id"] for c in copied}


def test_i04_an_import_opens_where_the_bundle_was_read_not_at_a_save_point(client):
    """A Save Point in the file is a position someone named, not the position the
    campaign is read at. The head comes from `headDepth`, as it did before M4."""
    _turns(client, 5)
    _undo(client)
    _undo(client)
    _save(client, "Way back at the start")  # made here, then the head moves on
    _redo(client)
    undone_story = _texts(client)

    imported = _import(client, _export(client))

    assert _story_of(imported["id"]) == undone_story
    assert _adventure(client, adv_id=imported["id"])["can_redo"] is True
    # And the Save Point arrived pointing somewhere else entirely.
    saved = _list(client, adv_id=imported["id"])[0]
    assert saved["depth"] < _head(imported["id"])[1]


def test_i04_an_imported_save_point_restores_in_the_new_campaign(client):
    _turns(client, 2)
    made = _save(client, "Before entering the abbey")
    at_save = _texts(client)
    _turns(client, 3)

    imported = _import(client, _export(client))
    copied = _list(client, adv_id=imported["id"])[0]
    assert copied["id"] != made["id"]

    r = _restore(client, copied["id"], adv_id=imported["id"])
    assert r.status_code == 200, r.text
    assert _story_of(imported["id"]) == at_save
    # The original campaign did not move.
    assert len(_texts(client)) == 11


def test_a_bundle_written_before_m4_imports_with_no_save_points(client):
    """Backward compatibility. A file with no `checkpoints` key is one written
    when Save Points did not exist, and a campaign that had none is what it
    records — so it opens, and it opens empty."""
    _turns(client, 3)
    told = _texts(client)
    bundle = _export(client)
    del bundle["checkpoints"]

    imported = _import(client, bundle)

    assert _story_of(imported["id"]) == told
    assert _list(client, adv_id=imported["id"]) == []


def test_a_save_point_naming_a_turn_the_file_does_not_carry_is_dropped(client):
    """A bookmark pointing outside the story is dropped rather than refusing the
    whole import. The head is checked the other way, because misplacing *it*
    affects every read in the file."""
    _turns(client, 2)
    _save(client, "Real")
    bundle = _export(client)
    bundle["checkpoints"].append(
        {"name": "Imaginary", "note": "", "branch": 0, "depth": 999}
    )
    bundle["checkpoints"].append({"name": "   ", "branch": 0, "depth": 1})
    bundle["checkpoints"].append({"name": "No branch", "branch": 77, "depth": 1})

    imported = _import(client, bundle)

    assert [c["name"] for c in _list(client, adv_id=imported["id"])] == ["Real"]


def test_save_points_on_two_branches_survive_the_round_trip(client):
    _turns(client, 2)
    shared = _save(client, "The fork")
    _turns(client, 2)
    old_line = _save(client, "Down the old road")

    _restore(client, shared["id"])
    ScriptedProvider.replies = ["Through the side door.\n```state\n{\"player.gold\": 1}\n```"]
    _play(client, "go around the back")
    new_line = _save(client, "Down the new road")

    imported = _import(client, _export(client))
    copied = _list(client, adv_id=imported["id"])

    assert sorted(c["name"] for c in copied) == [
        "Down the new road", "Down the old road", "The fork",
    ]
    # Three names, and they did not all collapse onto one branch.
    assert len({c["branch_id"] for c in copied}) == 2
    assert all(c["resolved"] for c in copied)
    del old_line, new_line


def _branch_rows(adv_id):
    db = SessionLocal()
    try:
        return db.query(models.Branch).filter_by(adventure_id=adv_id).all()
    finally:
        db.close()


# ------------------------------------------- L03: reconstruction after restart

def test_l03_a_save_point_restores_the_right_state_after_a_restart(client):
    _turns(client, 3)
    banked = _gold(client.adv_id)
    told = _texts(client)
    made = _save(client, "Before entering the abbey")

    _turns(client, 4)
    assert _gold(client.adv_id) == banked + 4 * GOLD_PER_TURN

    after = _restart(client)
    try:
        r = _restore(after, made["id"])
        assert r.status_code == 200, r.text
        assert _texts(after) == told
        assert _gold(after.adv_id) == banked
        # And the later history is still all there.
        assert len(_rows(after.adv_id)) == 1 + 2 * 7
    finally:
        after.close()


def test_l03_the_save_point_list_is_rebuilt_from_rows_alone(client):
    _turns(client, 2)
    first = _save(client, "One", note="a note")
    _turns(client, 2)
    second = _save(client, "Two")

    after = _restart(client)
    try:
        kept = _list(after)
        assert [c["name"] for c in kept] == ["Two", "One"]  # newest first
        assert kept[1]["note"] == "a note"
        assert [c["id"] for c in kept] == [second["id"], first["id"]]
        assert all(c["resolved"] for c in kept)
    finally:
        after.close()


# ---------------------------------------------------- the schema M4 adds

def test_an_m3_database_gains_the_save_point_table_and_keeps_its_story():
    """An M3 campaign database opens under M4 with no Save Points and no loss.

    The table is created by `create_all`, on existing databases as well as fresh
    ones, exactly as `memories` and `branches` were before it; migration 80 adds
    the index. What this proves is that a database stamped at M3's version — one
    that has never seen a `checkpoints` table — reaches M4's version with the
    table, the index, and every row it already had.
    """
    import os
    import tempfile
    from sqlalchemy import create_engine, inspect, text
    from sqlalchemy.orm import sessionmaker
    from app import migrations

    path = os.path.join(tempfile.mkdtemp(), "m3.db")
    m3 = create_engine(f"sqlite:///{path}")
    Base.metadata.create_all(bind=m3)
    # A campaign written by M3. Built through the models so that the columns it
    # carries are whatever the application writes, rather than a list this test
    # would have to keep current.
    session = sessionmaker(bind=m3)()
    try:
        owner = models.User(is_guest=False, email="m3@example.com")
        session.add(owner)
        session.flush()
        session.add(models.Adventure(user_id=owner.id, title="Old", head_depth=2))
        session.commit()
    finally:
        session.close()
    with m3.begin() as conn:
        # Now make it an M3 *schema*: no Save Points, stamped at M3's version.
        conn.execute(text("DROP TABLE checkpoints"))
        conn.execute(text("PRAGMA user_version = 79"))
    assert "checkpoints" not in inspect(m3).get_table_names()

    migrations.bootstrap(m3)

    insp = inspect(m3)
    assert "checkpoints" in insp.get_table_names()
    assert {c["name"] for c in insp.get_columns("checkpoints")} == {
        "id", "adventure_id", "name", "note", "branch_id", "depth",
        "created_at", "updated_at",
    }
    assert "ix_checkpoints_adventure" in {i["name"] for i in insp.get_indexes("checkpoints")}
    with m3.connect() as conn:
        assert conn.execute(text("PRAGMA user_version")).scalar() == 80
        # No Save Points were invented for a campaign whose owner named none...
        assert conn.execute(text("SELECT COUNT(*) FROM checkpoints")).scalar() == 0
        # ...and the campaign it already had is untouched.
        assert conn.execute(text("SELECT title, head_depth FROM adventures")).one() == ("Old", 2)

    # Running it again changes nothing.
    migrations.bootstrap(m3)
    assert "checkpoints" in inspect(m3).get_table_names()
    m3.dispose()


# ------------------------------------------- the cost of listing Save Points

def _sql_during(work):
    """Returns every SQL statement a block of work executed."""
    from sqlalchemy import event
    seen: list[str] = []

    def record(conn, cursor, statement, params, context, executemany):
        seen.append(statement)

    event.listen(engine, "before_cursor_execute", record)
    try:
        work()
    finally:
        event.remove(engine, "before_cursor_execute", record)
    return seen


def test_listing_save_points_costs_a_bounded_number_of_queries(client):
    """M4 review §R B-1. The list resolved each Save Point on its own, so the
    query count grew with the list: 53 SELECTs for 25 Save Points, against 4 for
    the comparable branch panel.

    The assertion is on *growth*, not on an exact number, because a fixed budget
    would be a number to edit rather than a rule to keep. Five times the Save
    Points must not mean five times the queries.
    """
    _turns(client, 5)
    for n in range(5):
        _save(client, f"Save Point {n}")
    few = _sql_during(lambda: _list(client))

    _turns(client, 20)
    for n in range(20):
        _save(client, f"Later Save Point {n}")
    assert len(_list(client)) == 25
    many = _sql_during(lambda: _list(client))

    # Five times the rows, and the query count does not move at all.
    assert len(many) == len(few), (
        f"listing 25 Save Points cost {len(many)} queries where 5 cost {len(few)}"
    )
    # And the whole thing is a handful, not a per-row walk.
    assert len(many) <= 6, f"{len(many)} queries to list 25 Save Points"


def test_listing_save_points_does_not_read_story_prose(client):
    """The other half of B-1. Resolving a coordinate asks whether a row exists;
    it never needs the narration in it, and `paging.py` states the rule this
    follows — a bulk read names the columns it needs.

    Enforced on the emitted SQL rather than on a byte count, because the failure
    this guards against is a `SELECT` widening back to the whole entity, which a
    small fixture would not make visible in bytes.
    """
    _turns(client, 3)
    for n in range(3):
        _save(client, f"Save Point {n}")

    statements = _sql_during(lambda: _list(client))
    action_reads = [q for q in statements if "FROM actions" in q]
    assert action_reads, "the list must still check that coordinates resolve"
    for query in action_reads:
        selected = query.split("FROM actions")[0]
        for column in ("actions.text", "actions.reasoning", "actions.world_delta"):
            assert column not in selected, f"{column} fetched to render the list:\n{query}"


def test_a_save_point_on_a_deleted_turn_is_still_reported_unresolved(client):
    """The bulk resolution must not have quietly changed what `resolved` means.

    This is the negative control for the B-1 rewrite: one query for many
    coordinates is only correct if a coordinate with no live row still comes
    back false.
    """
    _turns(client, 3)
    alive = _save(client, "Still here")
    _turns(client, 1)
    doomed = _save(client, "About to vanish")

    db = SessionLocal()
    try:
        for row in db.query(models.Action).filter_by(
            adventure_id=client.adv_id,
            branch_id=doomed["branch_id"],
            depth=doomed["depth"],
        ):
            db.delete(row)
        db.commit()
    finally:
        db.close()

    by_id = {c["id"]: c for c in _list(client)}
    assert by_id[doomed["id"]]["resolved"] is False
    assert by_id[alive["id"]]["resolved"] is True
    # One resolving and one not, in the same single query.


def test_the_bulk_resolution_does_not_confuse_coordinates_across_branches(client):
    """A coordinate is a pair, and the bulk query must match it as a pair.

    Matching `branch IN (...) AND depth IN (...)` would take the cross product,
    and a Save Point at a depth that exists on *another* branch would be
    reported as resolved. This builds exactly that trap: two branches, and a
    Save Point whose own coordinate is dead while the other branch has a live
    row at the same depth.
    """
    _turns(client, 4)
    _undo(client)
    _undo(client)
    ScriptedProvider.replies = ["A new road.\n```state\n{\"player.gold\": 1}\n```"]
    _play(client, "the other way")          # forks; new branch has rows at 5,6
    on_new_line = _save(client, "On the new line")

    # A Save Point on the old line at the same depth, whose row we then remove.
    db = SessionLocal()
    try:
        adventure = db.get(models.Adventure, client.adv_id)
        old_branch = (
            db.query(models.Branch)
            .filter_by(adventure_id=client.adv_id, parent_branch_id=None)
            .one()
        )
        stranded = models.Checkpoint(
            adventure_id=client.adv_id,
            name="Stranded on the old line",
            branch_id=old_branch.id,
            depth=on_new_line["depth"],
        )
        db.add(stranded)
        db.flush()
        stranded_id = stranded.id
        # Remove the old line's row at that depth, so this coordinate is dead
        # while the *other* branch still has a live row at the same depth.
        for row in db.query(models.Action).filter_by(
            adventure_id=client.adv_id,
            branch_id=old_branch.id,
            depth=on_new_line["depth"],
        ):
            db.delete(row)
        db.commit()
    finally:
        db.close()

    by_id = {c["id"]: c for c in _list(client)}
    assert by_id[on_new_line["id"]]["resolved"] is True
    assert by_id[stranded_id]["resolved"] is False, (
        "a dead coordinate was reported resolved because another branch has a "
        "live row at the same depth"
    )


# --------------------------------- deleting a branch, and saying so first

def test_the_branch_list_reports_how_many_save_points_each_line_carries(client):
    """The number the delete warning is built from (M4 review §R B-2).

    Served as part of the branch list rather than from an endpoint of its own,
    and as one grouped query rather than one per branch — the panel already
    reads this list to draw itself.
    """
    _turns(client, 2)
    _undo(client)
    _undo(client)
    ScriptedProvider.replies = ["A new road.\n```state\n{\"player.gold\": 1}\n```"]
    _play(client, "the other way")
    on_new = _save(client, "On the new line")

    branches = client.get(f"/api/adventures/{client.adv_id}/branches").json()
    by_id = {b["id"]: b for b in branches}
    assert len(branches) == 2
    assert by_id[on_new["branch_id"]]["save_points"] == 1
    other = next(b for b in branches if b["id"] != on_new["branch_id"])
    assert other["save_points"] == 0

    _save(client, "A second one here")
    branches = client.get(f"/api/adventures/{client.adv_id}/branches").json()
    assert {b["id"]: b["save_points"] for b in branches}[on_new["branch_id"]] == 2


def test_the_save_point_count_matches_what_blocks_the_deletion(client):
    """The number the panel disables its Delete button on has to be the same
    number the server refuses on, or the UI and the rule disagree."""
    _turns(client, 4)
    _undo(client)
    _undo(client)
    ScriptedProvider.replies = ["Second line.\n```state\n{\"player.gold\": 1}\n```"]
    _play(client, "second line")
    middle_branch = _head(client.adv_id)[0]
    _undo(client)
    ScriptedProvider.replies = ["Third line.\n```state\n{\"player.gold\": 1}\n```"]
    _play(client, "third line")
    _save(client, "Deep one")

    branches = client.get(f"/api/adventures/{client.adv_id}/branches").json()
    counts = {b["id"]: b["save_points"] for b in branches}
    # The count is per branch; the client sums it over the subtree, and the
    # server refuses on the same subtree.
    assert sum(counts.values()) == 1
    assert counts[middle_branch] == 0

    root_id = _root_branch(client.adv_id)
    client.post(f"/api/adventures/{client.adv_id}/branches/{root_id}/switch")
    assert client.delete(
        f"/api/adventures/{client.adv_id}/branches/{middle_branch}"
    ).status_code == 409


def test_both_branch_delete_surfaces_explain_the_save_point_rule():
    """The rule must be visible in every view the deletion is reachable from.

    A source-level assertion, because the project has no frontend test runner
    (M8). `test_offline_assets.py` reads the frontend the same way, for the same
    reason: the check is worth having now, and it is honest about what it is —
    it proves the wiring is in the build, not that a user saw it. The browser
    smoke test performed at closeout is what proves the rendering.

    Two files, because the branch list and the tree overlay each render their
    own delete control, and a rule that held in one of them would not be a rule.
    """
    from pathlib import Path

    repo = Path(__file__).resolve().parents[2]
    views = {
        "the branch panel":
            repo / "frontend/src/pages/Play/panels/BranchPanel.jsx",
        "the tree overlay":
            repo / "frontend/src/BranchMap.jsx",
    }
    for where, path in views.items():
        source = path.read_text(encoding="utf-8")
        assert "savePointsUnder" in source, (
            f"{where} does not count the Save Points that protect a branch"
        )
        # The Delete control is disabled while Save Points protect the subtree,
        # and says why rather than failing silently on the server.
        assert "protecting > 0" in source, (
            f"{where} does not disable Delete while Save Points protect the branch"
        )
        assert "deleting a Save Point deletes no story" in source, (
            f"{where} does not tell the user how to proceed"
        )
        # The user-facing copy must not explain itself in schema terms.
        for jargon in ("cascade", "foreign key", "foreign-key", "ON DELETE"):
            assert jargon.lower() not in source.lower(), (
                f"{where} uses implementation jargon in user-facing copy: {jargon}"
            )


def test_creating_a_save_point_takes_the_campaigns_turn_lock(client):
    """M4 review §S C-5. "Save where I am" has to name one committed position,
    and the head is exactly what a turn in flight is about to move.

    Asserted by holding the lock and watching Create refuse, which is the same
    contract Undo, Redo and Restore already answer with a 409. That proves it
    participates in the serialization rather than merely being fast.
    """
    _turns(client, 2)
    adventures.turns.acquire_turn_lock(client.adv_id)
    try:
        r = client.post(f"/api/adventures/{client.adv_id}/checkpoints",
                        json={"name": "During a turn"})
        assert r.status_code == 409, r.text
        # The same refusal the other position-moving operations give.
        assert _undo(client).status_code == 409
        assert _restore(client, 1).status_code in (404, 409)
    finally:
        adventures.turns._active_turns.discard(client.adv_id)

    # Nothing was written while the lock was held...
    assert _list(client) == []
    # ...and the endpoint works again once the turn is done.
    assert _save(client, "After the turn")["name"] == "After the turn"


def test_creating_a_save_point_releases_the_lock_even_when_it_refuses(client):
    """A refused create must not leave the campaign wedged.

    The empty-story refusal is raised from inside the locked section, so this is
    the case that would strand the lock if the release were not in a `finally`.
    """
    db = SessionLocal()
    try:
        adv = db.get(models.Adventure, client.adv_id)
        adv.head_depth = lineage.NO_DEPTH
        db.commit()
    finally:
        db.close()

    r = client.post(f"/api/adventures/{client.adv_id}/checkpoints",
                    json={"name": "Nowhere"})
    assert r.status_code == 400

    # The lock is free: an ordinary turn still works.
    assert client.adv_id not in adventures.turns._active_turns
    _play(client, "carry on")


def test_rename_and_delete_do_not_need_the_turn_lock(client):
    """Deliberate, and worth pinning down: neither reads nor moves a story
    position, so neither can race a turn in flight. Renaming a Save Point while
    a turn generates is a label edit, and refusing it would be a worse product
    for no safety gained."""
    _turns(client, 2)
    made = _save(client, "One")

    adventures.turns.acquire_turn_lock(client.adv_id)
    try:
        renamed = client.patch(
            f"/api/adventures/{client.adv_id}/checkpoints/{made['id']}",
            json={"name": "Renamed mid-turn"},
        )
        assert renamed.status_code == 200, renamed.text
        assert renamed.json()["name"] == "Renamed mid-turn"
        # And the coordinate did not move while a turn was in flight.
        assert (renamed.json()["branch_id"], renamed.json()["depth"]) == (
            made["branch_id"], made["depth"]
        )
        assert client.delete(
            f"/api/adventures/{client.adv_id}/checkpoints/{made['id']}"
        ).status_code == 204
    finally:
        adventures.turns._active_turns.discard(client.adv_id)