interactive-story/backend/app/narrative/store.py

"""M5: writing accepted state, atomically with the turn that caused it.

This is the only module in the package that touches the database, and the only
place authoritative narrative state is written.

## The atomicity rule (L01)

Everything a turn establishes goes in one transaction: the narration, the head
movement, the accepted events, the resulting snapshot, and the provenance. This
function *adds* to the caller's session and never commits — the turn engine's
single `db.commit()` remains the one commit point, so a failure anywhere before
it rolls the whole turn back rather than leaving narration accepted with half its
state written.

That ordering is deliberate and load-bearing. `L01` forbids a head position that
implies an accepted reply whose state commit did not complete, and the cheapest
way to guarantee that is to never have two commits to get out of step.

## What is not here

No reconstruction. Nothing in this module reads `state_events` to rebuild a
document — the snapshot on the node is the restore path
(`TECHNICAL-DESIGN.md` §10.4). The events are the audit trail, and an audit
trail that the system depends on for correctness stops being an audit trail and
becomes a replay engine.
"""

from __future__ import annotations

import copy

from sqlalchemy.orm import Session

from .. import models
from . import apply as apply_module
from . import model


def current(adventure: models.Adventure) -> dict:
    """The campaign's authoritative state right now, as a document.

    Normalised on the way out, so every caller gets the same shape whatever a
    hand-edited row or an older snapshot contains.
    """
    return model.normalize(adventure.narrative_state)


def set_current(adventure: models.Adventure, state: dict) -> None:
    adventure.narrative_state = model.normalize(state)


def canon_of(adventure: models.Adventure) -> dict:
    """The campaign's own rules, which outrank anything a narration proposes.

    Configuration rather than code (C01, J03): the campaign says what it forbids,
    and `validate` enforces it without knowing what the rule means.
    """
    canon = adventure.campaign_canon
    return canon if isinstance(canon, dict) else {}


def record(
    db: Session,
    adventure: models.Adventure,
    *,
    review,
    raw_block: str = "",
    parsed=None,
    action: models.Action | None = None,
    branch_id: int | None = None,
    depth: int | None = None,
    model_name: str = "",
    source: str = "accepted_story",
) -> tuple[dict, models.StateProposal]:
    """Applies a reviewed proposal and records everything about it.

    Returns `(new_state, proposal_row)`. The caller is responsible for putting
    the new state where it belongs — on the campaign, and on the node's snapshot
    — because only the caller knows whether this is a turn, a retry or a
    correction.

    Nothing is committed here. See the module docstring.
    """
    before = current(adventure)
    after = apply_module.apply_events(
        before, review.accepted, branch_id=branch_id, depth=depth, source=source
    )

    proposal = models.StateProposal(
        adventure_id=adventure.id,
        action_id=action.id if action is not None else None,
        branch_id=branch_id,
        depth=depth,
        model_name=model_name or "",
        source=source,
        status=review.status,
        raw_output=raw_block or "",
        detail={
            "parsed": parsed,
            "accepted": review.accepted,
            "rejected": [r.as_dict() for r in review.rejected],
        },
    )
    db.add(proposal)
    # The proposal needs an id before its events can point at it, and the
    # session does not autoflush. This is a flush, not a commit: still one
    # transaction, still all-or-nothing.
    db.flush()

    for sequence, event in enumerate(review.accepted):
        db.add(models.StateEvent(
            adventure_id=adventure.id,
            proposal_id=proposal.id,
            action_id=action.id if action is not None else None,
            branch_id=branch_id,
            depth=depth,
            sequence=sequence,
            event_type=event.get("type", ""),
            payload=copy.deepcopy(event),
            before=_before_value(before, event),
            source=source,
        ))
    return after, proposal


def _before_value(state: dict, event: dict) -> dict | None:
    """What the value this event changes was, immediately beforehand.

    Recorded per event so §8's "what was the previous value" is answerable
    without replaying anything. Only the slice the event touches: a whole
    document per event would duplicate the snapshot for no extra answer.
    """
    kind = event.get("type")
    if kind in ("set_entity_status", "set_entity_attribute",
                "set_entity_conditions", "set_current_location"):
        entity = model.entity(state, event.get("entity", ""))
        if entity is None:
            return None
        if kind == "set_entity_status":
            return {"status": entity.get("status")}
        if kind == "set_entity_attribute":
            attribute = event.get("attribute")
            return {"attribute": attribute,
                    "value": (entity.get("attributes") or {}).get(attribute)}
        if kind == "set_entity_conditions":
            return {"conditions": list(entity.get("conditions") or [])}
        return {"location": entity.get("location")}
    if kind in ("set_possession", "clear_possession"):
        return {"owner": model.owner_of(state, event.get("item", ""))}
    if kind == "invalidate_fact":
        for fact in state.get("facts") or []:
            if fact.get("id") == event.get("fact_id"):
                return {"status": fact.get("status"), "predicate": fact.get("predicate")}
        return None
    if kind == "resolve_story_thread":
        thread = (state.get("threads") or {}).get(event.get("thread", ""))
        return {"status": thread.get("status")} if isinstance(thread, dict) else None
    if kind == "end_relationship":
        return {"status": "active"}
    return None


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

def events_for(
    db: Session, adventure: models.Adventure, action_id: int
) -> list[models.StateEvent]:
    """The accepted events one node's narration produced, in order."""
    return (
        db.query(models.StateEvent)
        .filter(
            models.StateEvent.adventure_id == adventure.id,
            models.StateEvent.action_id == action_id,
        )
        .order_by(models.StateEvent.sequence, models.StateEvent.id)
        .all()
    )


def history(
    db: Session, adventure: models.Adventure, limit: int = 200
) -> list[models.StateEvent]:
    """The campaign's accepted state events, newest first.

    Bounded by default: this is an audit view, and an unbounded read of a long
    campaign's every event is the kind of query this project keeps a regression
    test about.
    """
    return (
        db.query(models.StateEvent)
        .filter(models.StateEvent.adventure_id == adventure.id)
        .order_by(models.StateEvent.id.desc())
        .limit(limit)
        .all()
    )