interactive-story/backend/app/head.py

"""M3: where the story is being read, and what moving that point costs.

Undo used to delete. It removed the trailing nodes, let `tree.refresh_head`
recompute the tip from what survived, and the story was wherever the rows ended.
That made the head a derived value and made Redo impossible, because the turns it
would have moved forward into were gone.

The head is now a stored position that can sit behind the retained tip. Nothing
is deleted, so three things that used to be the same question are now three
different ones:

* **the active head** — `adventure.head_branch_id` and `adventure.head_depth`,
  the end of the story being told. Every read of the story stops here, because
  `lineage.Path` caps every entry at it.
* **the retained tip** — the deepest live node still on the lineage. Redo walks
  toward it. It is read through `Path.uncapped()`, and only this module and the
  divergence check may look at it.
* **the opening** — the shallowest node on the story, which is the floor Undo
  may not pass.

Everything that moves the head or asks a question about it lives here, so the
turn engine, Retry, Add-take, Undo, Redo and Edit share one set of rules rather
than four similar ones. The Phase 0B spike put the fork check in the write path
and left Retry and Add-take on the old one, which is exactly the divergence this
module exists to prevent.

The state that belongs to a position is not recomputed. Every node carries the
world state it left behind (`attempts.snapshot_outcome`), so moving the head is a
row lookup plus `attempts.restore_state`, at any distance, in either direction.
"""

from sqlalchemy.orm import Session, undefer

from . import attempts, models, tree
from .context import lineage

# The kinds of node a player writes. An undo or a redo steps over a whole turn,
# which is one of these followed by the reply to it, so both ends need to agree
# on what "a player's half of a turn" is.
PLAYER_TYPES = ("do", "say", "story", "continue")


# ------------------------------------------------------------------ reading

def opening_depth(db: Session, adventure: models.Adventure) -> int | None:
    """Returns the depth of the first node of the story, or None if there is none.

    This is Undo's floor. The Phase 0B spike moved the head to -1 and rendered an
    empty transcript, because its guard tested for a node of type `start` and an
    adventure opened with a player-written `story` action has none. Asking the
    path for its shallowest node needs no such special case: whatever the opening
    is called, it is the node with the smallest depth, and the story keeps it.

    The read is uncapped. The opening does not move when the head does, and
    capping would make the floor depend on where the head already is.
    """
    return (
        db.query(models.Action.depth)
        .filter(
            models.Action.adventure_id == adventure.id,
            lineage.path_of(db, adventure).uncapped().clause(models.Action),
        )
        .order_by(models.Action.depth.asc(), models.Action.id.asc())
        .limit(1)
        .scalar()
    )


def retained_tip(db: Session, adventure: models.Adventure) -> int | None:
    """Returns the depth of the deepest live node still retained on this lineage.

    This is what the head would be if the story had never been undone, and it is
    what Redo can reach. It is not the head, and no read of the story may use it.
    """
    return (
        db.query(models.Action.depth)
        .filter(
            models.Action.adventure_id == adventure.id,
            lineage.path_of(db, adventure).uncapped().clause(models.Action),
        )
        .order_by(models.Action.depth.desc(), models.Action.id.desc())
        .limit(1)
        .scalar()
    )


def behind_tip(db: Session, adventure: models.Adventure) -> bool:
    """Returns whether retained story sits past the head.

    This one predicate answers every "does this write need to fork?" question in
    the application. It is true exactly when the user has undone and not redone,
    which is the only situation in which writing can displace an accepted future.

    It is also what makes "is this turn the tip?" answerable again. Retry,
    Add-take and `stand_on` each decide between amending a turn in place and
    giving it a branch, and each used to ask `last_action`, which reads the
    *capped* path and therefore reports the node at the head as the newest one.
    Under a moved-back head that answer is wrong in the dangerous direction: it
    says a turn with an accepted future is a leaf, and amending it in place would
    leave that future descending from a take that is no longer live.
    """
    tip = retained_tip(db, adventure)
    return tip is not None and tip > adventure.head_depth


def node_at(
    db: Session, adventure: models.Adventure, depth: int
) -> models.Action | None:
    """Returns the live node at `depth` on the retained lineage, outcome loaded.

    The read is uncapped on purpose: Redo asks for a node it is about to move the
    head onto, which is by definition past the head at the time of asking. The
    outcome columns are undeferred because the only reason to fetch this row is
    to restore the state it left behind.
    """
    return (
        db.query(models.Action)
        .filter(
            models.Action.adventure_id == adventure.id,
            lineage.path_of(db, adventure).uncapped().clause(models.Action),
            models.Action.depth == depth,
        )
        .options(
            undefer(models.Action.state_after),
            undefer(models.Action.world_state_after),
        )
        .order_by(models.Action.id)
        .first()
    )


def redo_target(db: Session, adventure: models.Adventure) -> int | None:
    """Returns the depth the head moves to on Redo, or None if there is nowhere.

    Redo steps over a whole turn, the same unit Undo steps back over, so a
    player's action and the reply to it move together. Landing between them would
    show the story an input with no answer and would leave the next Undo undoing
    half a turn.

    The walk is along the retained lineage, which is what makes Redo follow the
    continuation that was active rather than choosing among branches. After a
    divergence the new branch *is* the lineage, and the displaced future is no
    longer on it, so this returns None without having to know that a divergence
    happened. That is `STORY-BRANCH-SEMANTICS.md` §8 falling out of the lineage
    rather than being enforced by a flag.
    """
    ahead = (
        db.query(models.Action)
        .filter(
            models.Action.adventure_id == adventure.id,
            lineage.path_of(db, adventure).uncapped().clause(models.Action),
            models.Action.depth > adventure.head_depth,
        )
        .order_by(models.Action.depth.asc(), models.Action.id.asc())
        .limit(2)
        .all()
    )
    if not ahead:
        return None
    first = ahead[0]
    if (
        first.type in PLAYER_TYPES
        and len(ahead) > 1
        and ahead[1].type == "ai"
        and ahead[1].depth == (first.depth or 0) + 1
    ):
        return ahead[1].depth
    return first.depth


def can_redo(db: Session, adventure: models.Adventure) -> bool:
    """Returns whether an ordinary Redo is available from where the head is."""
    return redo_target(db, adventure) is not None


def undo_target(
    db: Session, adventure: models.Adventure
) -> tuple[int, models.Action] | None:
    """Returns where Undo moves the head, and the first node it steps back over.

    None means there is nothing to undo, which is either an empty story or a head
    already resting on the opening. The caller turns that into a 400; this
    function does not raise, so that the same question can be asked without
    committing to undoing.

    A turn is the player's node plus the reply to it, and both move together for
    the reason given in `redo_target`. The player half is only claimed when it is
    directly in front of the reply, so a bare `continue`, which writes no player
    node, steps back over the reply alone.
    """
    newest = (
        db.query(models.Action)
        .filter(
            models.Action.adventure_id == adventure.id,
            lineage.path_of(db, adventure).clause(models.Action),
        )
        .order_by(models.Action.depth.desc(), models.Action.id.desc())
        .limit(2)
        .all()
    )
    if not newest:
        return None
    last = newest[0]
    first_stepped = last
    before = newest[1] if len(newest) > 1 else None
    if (
        last.type == "ai"
        and before is not None
        and before.type in PLAYER_TYPES
        and before.depth == (last.depth or 0) - 1
    ):
        first_stepped = before
    floor = opening_depth(db, adventure)
    if first_stepped.depth is None or floor is None:
        return None
    if first_stepped.depth <= floor:
        # Stepping back over this turn would hide the opening of the campaign,
        # which is the pre-campaign state `STORY-BRANCH-SEMANTICS.md` §4 stops
        # at. The floor is the opening node itself rather than depth -1, so an
        # adventure that opens on a player-written `story` action stops in the
        # same place as one that opens on a `start` node.
        return None
    return first_stepped.depth - 1, first_stepped


def can_undo(db: Session, adventure: models.Adventure) -> bool:
    """Returns whether an ordinary Undo is available from where the head is."""
    return undo_target(db, adventure) is not None


def displaced_history_under(
    db: Session, adventure: models.Adventure, node: models.Action
) -> bool:
    """Returns whether story the reader cannot see descends from `node`.

    This is the question an in-place edit has to ask. Editing rewrites one row
    and re-evaluates nothing, which is what makes it a correction rather than a
    new continuation. That is harmless while everything descending from the row
    is on screen: the reader can see what their correction has to stay
    consistent with. It stops being harmless the moment a continuation descends
    from the row and is *not* on screen, because the edit then silently changes
    the words an invisible stretch of story was written from. That is the one
    way M3's retained history can be made to contradict itself.

    Refusing is deliberately the whole of the fix. Making such an edit fork, so
    the original text and its future stay whole, is
    `STORY-BRANCH-SEMANTICS.md` §14-15 — and §15 requires re-evaluating the
    state the edited prose implies, which is M5's extraction pass. Neither is
    started here.

    The question is asked as one shape rather than two, because the two ways a
    descendant becomes invisible turn out to be the same fact. An undone future
    sits past the head on this very lineage; a displaced line sits past a fork
    on a branch the story left. In both cases there is a live node, deeper than
    this one, that descends from it and is not on the path being read — and the
    departed branch is usually an *ancestor* of the branch now being read, which
    is why "branches other than the active one" is the wrong set to look at.

    Only the deepest live node on each descending branch is examined. Whether a
    node is on the read path is monotone in depth: a branch is on the path with
    a cap, and a node is visible when its depth is at or under that cap. So if
    the deepest one is visible, every shallower one is too, and if it is not,
    the answer is already yes.

    A node that is not live has no descendants of its own — a take the story
    moved past keeps a continuation only by being forked, and that fork is a
    branch this loop asks about anyway — so editing one is always safe.
    """
    if not node.live or node.depth is None:
        return False
    read = lineage.path_of(db, adventure)
    branches = (
        db.query(models.Branch)
        .filter(models.Branch.adventure_id == adventure.id)
        .all()
    )
    for branch in branches:
        # Uncapped: the question is what this branch's story descends from, not
        # how much of it the reader is currently being shown.
        if not lineage.Path(lineage.entries_of(branch)).contains(node):
            continue
        deepest = (
            db.query(models.Action)
            .filter(
                models.Action.adventure_id == adventure.id,
                models.Action.branch_id == branch.id,
                models.Action.live.is_(True),
                models.Action.depth > node.depth,
            )
            .order_by(models.Action.depth.desc(), models.Action.id.desc())
            .first()
        )
        if deepest is not None and not read.contains(deepest):
            return True
    return False


# ------------------------------------------------------------------ writing

def move_to(db: Session, adventure: models.Adventure, depth: int) -> None:
    """Moves the active head to `depth` and restores the state recorded there.

    This is the whole of Undo and Redo. Nothing is deleted, nothing is
    recomputed, and the direction of travel does not matter: the node at the
    destination carries the world state it left behind, so arriving from in front
    of it and arriving from behind it restore the same value.

    A destination with no node — the head resting one step in front of the
    opening — leaves the live state alone, which is `attempts.restore_state`'s
    rule for a missing snapshot and the reason it is not this function's job to
    invent an empty one.
    """
    adventure.head_depth = depth
    attempts.restore_state(adventure, node_at(db, adventure, depth))


def fork_if_behind_head(db: Session, adventure: models.Adventure) -> bool:
    """Gives the story a new branch when a write would displace a retained future.

    Returns whether a branch was created, which is what a caller reports as a
    divergence.

    Called before every write that continues the story, and it does nothing on
    the ordinary path where the head is already at the tip. That is the property
    worth keeping: a story that is never undone forks exactly as often as it did
    before M3, so the branch table does not fill up with one branch per turn.

    Undo alone must not fork. Moving the head is not a decision to abandon
    anything — the user may be reading, or about to Redo. Only the first write
    below the head states which continuation they mean, which is
    `STORY-BRANCH-SEMANTICS.md` §8 and §20 and what makes Redo survive an Undo.

    `tree.branch_at` leaves the departed branch exactly as it is: its nodes stay
    live, at their depths, on their branch. The new branch inherits the story up
    to the head and owns everything written from here, so the displaced future
    remains reachable through the branch it was written on.
    """
    if not behind_tip(db, adventure):
        return False
    departed = lineage.branch_of(db, adventure)
    at_depth = adventure.head_depth
    tree.branch_at(db, adventure, at_depth)
    if departed is not None:
        mark_superseded(departed, at_depth)
    return True


def mark_superseded(branch: models.Branch, depth: int) -> None:
    """Records that this branch's story past `depth` was displaced.

    `DATA-MODEL.md` §5 gives a branch a disposition of active, retained or
    disposable. This is that disposition, stored as the fact that produced it
    rather than as a word: the depth the story left at, and when. A branch with
    no `superseded_at` is active; one with a value has retained history past that
    depth which no active head is reading.

    Nothing in the application reads these columns to make a decision, and that
    is deliberate. Redo is decided by the lineage, not by a flag, so a stale or
    hand-edited value here cannot make the story wrong. They exist so that the
    cleanup and discarded-history features `STORY-BRANCH-SEMANTICS.md` §28 and
    §29 leave to a later version have something to select on, and so that a
    divergence is observable in a test.

    The shallowest departure wins. A branch left at depth 9 and later left again
    at depth 4 has retained history from 4 onward, and recording the later, deeper
    value would understate what was displaced.
    """
    if branch.superseded_depth is None or depth < branch.superseded_depth:
        branch.superseded_depth = depth
    if branch.superseded_at is None:
        branch.superseded_at = models.utcnow()