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

"""M5: turning accepted events into a new state document.

Pure and total. Every function here takes a document and returns a new one; none
touches the database, and none can fail on an event `validate.review` accepted —
validation is the only place an event is refused, so this module never has to
decide anything twice.

The dispatch is an explicit `if/elif` chain over `events.SPECS`, not a lookup
table keyed on the payload. The difference matters: a table maps a string a model
supplied to a callable, and the security of that arrangement rests entirely on
the allowlist being correct. A chain of literal comparisons cannot be steered by
a payload at all, whatever the allowlist does.
"""

from __future__ import annotations

import copy

from . import model


def apply_events(
    state: dict,
    accepted: list[dict],
    *,
    branch_id: int | None = None,
    depth: int | None = None,
    source: str = "accepted_story",
) -> dict:
    """Returns `state` with every event in `accepted` applied, in order.

    The input document is never mutated: head movement stores snapshots by
    reference in places, and a mutation here would edit the past.

    `branch_id`/`depth` stamp facts and relationships with where they were
    established, which is what makes the audit trail answer "which turn caused
    this" without a join. `source` records whether the campaign, the story or
    the user established it — C04's provenance, carried on the value itself.
    """
    document = model.normalize(state)
    for event in accepted:
        _apply_one(document, event, branch_id, depth, source)
    return document


def _apply_one(state: dict, event: dict, branch_id, depth, source: str) -> None:
    kind = event["type"]

    if kind == "create_entity":
        state["entities"][event["entity"]] = model.new_entity(
            type=event.get("entity_type") or "other",
            name=event["name"],
            description=event.get("description") or "",
            aliases=event.get("aliases") or [],
        )

    elif kind == "set_entity_status":
        _entity(state, event["entity"])["status"] = event["status"]

    elif kind == "set_entity_attribute":
        # Absolute assignment. The whole reason ADR 010 exists.
        _entity(state, event["entity"])["attributes"][event["attribute"]] = event["value"]

    elif kind == "set_entity_conditions":
        _entity(state, event["entity"])["conditions"] = list(event["conditions"])

    elif kind == "set_current_location":
        _entity(state, event["entity"])["location"] = event["location"]

    elif kind == "set_possession":
        state["possessions"][event["item"]] = event["owner"]

    elif kind == "clear_possession":
        state["possessions"].pop(event["item"], None)

    elif kind == "add_fact":
        state["facts"].append({
            "id": event.get("fact_id") or _fact_id(state),
            "subject": event.get("subject"),
            "predicate": event["predicate"],
            "object": event.get("object"),
            "value": event.get("value"),
            "authority": _authority(source),
            "source": source,
            "status": "active",
            "branch_id": branch_id,
            "depth": depth,
        })

    elif kind == "invalidate_fact":
        for fact in state["facts"]:
            if fact.get("id") == event["fact_id"]:
                # Withdrawn, not removed: C04 needs the record of what the
                # campaign used to believe, and a deleted row audits nothing.
                fact["status"] = "invalidated"
                fact["invalidated_by"] = source
                fact["invalidated_at"] = {"branch_id": branch_id, "depth": depth}
                if event.get("reason"):
                    fact["invalidated_reason"] = event["reason"]

    elif kind == "add_relationship":
        state["relationships"].append({
            "id": _relationship_id(state),
            "source": event["source"],
            "target": event["target"],
            "type": event["relationship"],
            "description": event.get("description") or "",
            "status": "active",
            "established_by": source,
            "branch_id": branch_id,
            "depth": depth,
        })

    elif kind == "end_relationship":
        for relationship in state["relationships"]:
            if (
                relationship.get("source") == event["source"]
                and relationship.get("target") == event["target"]
                and relationship.get("type") == event["relationship"]
                and relationship.get("status") == "active"
            ):
                relationship["status"] = "ended"
                relationship["ended_at"] = {"branch_id": branch_id, "depth": depth}

    elif kind == "open_story_thread":
        state["threads"][event["thread"]] = {
            "title": event["title"],
            "description": event.get("description") or "",
            "status": "open",
            "opened_at": {"branch_id": branch_id, "depth": depth},
        }

    elif kind == "resolve_story_thread":
        thread = state["threads"].get(event["thread"])
        if isinstance(thread, dict):
            thread["status"] = "resolved"
            thread["resolution"] = event.get("resolution") or ""
            thread["resolved_at"] = {"branch_id": branch_id, "depth": depth}

    elif kind == "set_scene":
        scene = dict(state.get("scene") or {})
        if "summary" in event:
            scene["summary"] = event["summary"]
        if "location" in event:
            scene["location"] = event["location"]
        if "present" in event:
            scene["present"] = list(event["present"] or [])
        scene["at"] = {"branch_id": branch_id, "depth": depth}
        state["scene"] = scene

    # No `else`. Every allowed type is handled above, and an unhandled one
    # cannot arrive: `validate.review` refuses anything outside the allowlist,
    # and the allowlist is this list. A silent fall-through would be the one way
    # an event could appear accepted and do nothing.


def _entity(state: dict, key: str) -> dict:
    """The entity record for `key`, created bare if a snapshot lost it.

    Validation guarantees the entity exists, so this is a repair path for a
    hand-edited or partially imported document rather than a normal branch. A
    bare record is better than a KeyError: the story is still readable, and the
    inspector shows an entity with nothing known about it, which is true.
    """
    entities = state["entities"]
    found = entities.get(key)
    if not isinstance(found, dict):
        found = model.new_entity(name=key)
        entities[key] = found
    found.setdefault("attributes", {})
    found.setdefault("conditions", [])
    return found


def _authority(source: str) -> str:
    """Which authority band a source's assertions carry.

    A user's correction outranks the story (C04); the story outranks a guess.
    `DATA-MODEL.md` §14 orders the bands, and this is the mapping into them.
    """
    if source == "manual_correction":
        return "manual_correction"
    if source == "campaign_canon":
        return "campaign_canon"
    return "accepted_story"


def _fact_id(state: dict) -> str:
    return f"f{len(state['facts']) + 1}"


def _relationship_id(state: dict) -> str:
    return f"r{len(state['relationships']) + 1}"


def diff(before: dict, after: dict) -> list[str]:
    """A short human-readable list of what changed between two documents.

    Shown under a turn the way the world-state chip used to be, and recorded on
    the node for the bulk read. Text rather than structure, because its only
    consumer is a person reading "Aldric now holds the silver key".
    """
    before = model.normalize(before)
    after = model.normalize(after)
    lines: list[str] = []

    for key, entity in after["entities"].items():
        was = before["entities"].get(key)
        name = model.entity_name(after, key)
        if was is None:
            lines.append(f"{name} enters the story")
            continue
        if was.get("status") != entity.get("status"):
            lines.append(f"{name} is now {entity.get('status')}")
        if was.get("location") != entity.get("location") and entity.get("location"):
            lines.append(f"{name} is at {model.entity_name(after, entity['location'])}")
        if sorted(was.get("conditions") or []) != sorted(entity.get("conditions") or []):
            now = ", ".join(entity.get("conditions") or []) or "nothing"
            lines.append(f"{name}: {now}")
        for attribute, value in (entity.get("attributes") or {}).items():
            if (was.get("attributes") or {}).get(attribute) != value:
                lines.append(f"{name} {attribute} = {value}")

    for item, owner in after["possessions"].items():
        if before["possessions"].get(item) != owner:
            lines.append(
                f"{model.entity_name(after, item)} → {model.entity_name(after, owner)}"
            )
    for item in before["possessions"]:
        if item not in after["possessions"]:
            lines.append(f"{model.entity_name(after, item)} is held by nobody")

    known = {f.get("id") for f in before["facts"]}
    for fact in after["facts"]:
        if fact.get("id") not in known:
            lines.append(f"fact: {_fact_text(after, fact)}")
    was_active = {f["id"] for f in model.active_facts(before)}
    for fact in before["facts"]:
        if fact.get("id") in was_active and fact.get("id") not in {
            f["id"] for f in model.active_facts(after)
        }:
            lines.append(f"withdrawn: {_fact_text(after, fact)}")

    known = {r.get("id") for r in before["relationships"]}
    for relationship in after["relationships"]:
        if relationship.get("id") not in known:
            lines.append(
                f"{model.entity_name(after, relationship['source'])} "
                f"{relationship['type']} "
                f"{model.entity_name(after, relationship['target'])}"
            )

    for key, thread in after["threads"].items():
        was = before["threads"].get(key)
        if was is None:
            lines.append(f"opened: {thread.get('title', key)}")
        elif was.get("status") != thread.get("status"):
            lines.append(f"{thread.get('status')}: {thread.get('title', key)}")

    return lines


def _fact_text(state: dict, fact: dict) -> str:
    parts = []
    if fact.get("subject"):
        parts.append(model.entity_name(state, fact["subject"]))
    parts.append(str(fact.get("predicate", "")))
    if fact.get("object"):
        parts.append(model.entity_name(state, fact["object"]))
    if fact.get("value") is not None:
        parts.append(str(fact["value"]))
    return " ".join(p for p in parts if p)