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

"""M5: deciding which proposed events the application will accept.

A proposal is untrusted model output. This module is the gate between it and the
authoritative state, and it is layered so that a rejection can say *which* rule
refused and a test can aim at one layer at a time:

    1. envelope    is this a proposal at all — a dict with a list of events?
    2. allowlist   is each event type one this application implements? (H05)
    3. schema      are the required fields present, and the right shape?
    4. referential do the entities and threads it names exist?
    5. semantic    does it contradict campaign canon, or itself?

Layer 2 is the security boundary and runs before any field is read, so a payload
carrying `command` or `path` alongside an unknown type is discarded without those
fields ever being looked at.

## What rejection means

Nothing is partially applied. `review` returns accepted and rejected events
separately and the caller decides; `apply.py` is only ever handed the accepted
list. A proposal with one bad event out of four therefore lands three, which is
`partially_accepted` — the alternative, discarding all four because the model
misspelled one entity, loses story the user watched happen.

What is *never* allowed is a rejected event mutating anything, or a rejection
being silent: every refusal carries a reason, is counted, and is stored on the
proposal record for §8's audit.

## What this module does not do

It does not decide whether the model was *right*. A typed event can be
well-formed, reference real entities, contradict nothing, and still describe
something the narration did not say. That is C06's territory and no validator
can settle it — ADR 010 says so plainly. What validation buys is that a wrong
proposal is wrong in a way a person can see in the audit trail, rather than one
that silently means something other than it appears to.
"""

from __future__ import annotations

from . import events, model

# A rejected event carries one of these, so tests and the debug view can assert
# on the reason rather than on prose.
UNKNOWN_TYPE = "unknown_event_type"
NOT_AN_OBJECT = "not_an_object"
MISSING_FIELD = "missing_field"
BAD_FIELD_TYPE = "bad_field_type"
UNKNOWN_REFERENCE = "unknown_reference"
CANON_CONFLICT = "canon_conflict"
SELF_CONTRADICTION = "self_contradiction"
DUPLICATE_ENTITY = "duplicate_entity"

# How many events one proposal may carry. A narration describes a turn, not a
# migration; a hundred events is a runaway model or a payload trying to be
# something else, and either way the cap bounds the work before it is done.
MAX_EVENTS = 40
# How long a text field may be. Long enough for a description, short enough that
# a proposal cannot smuggle a document into the state.
MAX_TEXT = 2_000
MAX_LABELS = 40


class Rejection:
    """One event that will not be applied, and why."""

    __slots__ = ("event", "reason", "detail")

    def __init__(self, event, reason: str, detail: str = ""):
        self.event = event
        self.reason = reason
        self.detail = detail

    def as_dict(self) -> dict:
        return {"event": self.event, "reason": self.reason, "detail": self.detail}

    def __repr__(self) -> str:  # pragma: no cover - debugging aid
        return f"<Rejection {self.reason}: {self.detail}>"


class Review:
    """The verdict on one proposal."""

    __slots__ = ("accepted", "rejected")

    def __init__(self, accepted: list[dict], rejected: list[Rejection]):
        self.accepted = accepted
        self.rejected = rejected

    @property
    def status(self) -> str:
        """`DATA-MODEL.md` §19's validation_status."""
        if self.rejected and self.accepted:
            return "partially_accepted"
        if self.rejected:
            return "rejected"
        return "accepted"

    def as_dict(self) -> dict:
        return {
            "status": self.status,
            "accepted": self.accepted,
            "rejected": [r.as_dict() for r in self.rejected],
        }


def review(payload, state: dict, canon: dict | None = None) -> Review:
    """Returns which of `payload`'s events may be applied to `state`.

    `state` is the document the events would apply to, needed because
    referential checks ask what already exists. `canon` carries the campaign's
    own rules, which outrank anything a narration proposes (C01).

    The state is **not** mutated. Events are checked against a running view that
    accounts for entities earlier events in the same proposal create, so a
    proposal may introduce Mara and then move her, but nothing is written until
    the caller applies the accepted list.
    """
    accepted: list[dict] = []
    rejected: list[Rejection] = []

    proposed = _events_of(payload)
    if proposed is None:
        return Review([], [Rejection(payload, NOT_AN_OBJECT,
                                     "the proposal is not an object with an event list")])

    # Entities this proposal has introduced, so a later event in the same
    # proposal may refer to them. Kept separately from `state` so that a
    # rejected create cannot make a later reference resolve.
    introduced: set[str] = set()

    for raw in proposed[:MAX_EVENTS]:
        problem = _check(raw, state, introduced, canon)
        if problem is not None:
            rejected.append(problem)
            continue
        accepted.append(raw)
        spec = events.spec(raw["type"])
        if spec and spec["creates"]:
            introduced.add(str(raw[spec["creates"]]))

    for extra in proposed[MAX_EVENTS:]:
        rejected.append(Rejection(extra, BAD_FIELD_TYPE,
                                  f"more than {MAX_EVENTS} events in one proposal"))
    return Review(accepted, rejected)


def _events_of(payload) -> list | None:
    """The event list, from either shape a proposal may legitimately take."""
    if isinstance(payload, list):
        return [e for e in payload]
    if not isinstance(payload, dict):
        return None
    found = payload.get("events")
    if found is None:
        return []
    if not isinstance(found, list):
        return None
    return found


def _check(raw, state: dict, introduced: set[str], canon: dict | None) -> Rejection | None:
    """Returns why `raw` is unacceptable, or None if it may be applied."""
    # ---- layer 1: is it an event-shaped object at all ----
    if not isinstance(raw, dict):
        return Rejection(raw, NOT_AN_OBJECT, "event is not an object")

    # ---- layer 2: the allowlist, before any field is read ----
    #
    # H05 lands here. `execute_shell` is refused because it is not in the
    # vocabulary, and its `command` field is never looked at — there is no
    # branch in this application that could reach it.
    event_type = raw.get("type", raw.get("event_type"))
    if not events.is_allowed(event_type):
        return Rejection(raw, UNKNOWN_TYPE, f"{event_type!r} is not a state event")
    raw["type"] = event_type
    spec = events.spec(event_type)

    # ---- layer 3: schema ----
    for field, kind in spec["required"].items():
        if field not in raw:
            return Rejection(raw, MISSING_FIELD, f"{event_type} needs {field!r}")
        bad = _bad_shape(raw[field], kind, field)
        if bad:
            return Rejection(raw, BAD_FIELD_TYPE, bad)
    for field, kind in spec["optional"].items():
        if field in raw and raw[field] is not None:
            bad = _bad_shape(raw[field], kind, field)
            if bad:
                return Rejection(raw, BAD_FIELD_TYPE, bad)

    # ---- layer 4: referential integrity ----
    known = set(state.get("entities") or {}) | introduced
    for field in spec["refs"]:
        named = raw.get(field)
        if named is None or field not in raw:
            continue          # optional reference, absent
        if not isinstance(named, str) or named not in known:
            return Rejection(raw, UNKNOWN_REFERENCE,
                             f"{event_type} names {field}={named!r}, which does not exist")
    if event_type == "add_fact" and raw.get("object") is not None:
        # An object may name an entity or another fact. Checking both keeps the
        # reference meaningful — a typo is still caught — without forcing every
        # thing a fact can be about to be promoted to an entity first.
        known_facts = {f.get("id") for f in (state.get("facts") or [])}
        target = raw["object"]
        if not isinstance(target, str) or (target not in known and target not in known_facts):
            return Rejection(raw, UNKNOWN_REFERENCE,
                             f"add_fact names object={target!r}, which does not exist")
    if event_type == "invalidate_fact":
        if not any(f.get("id") == raw["fact_id"] for f in (state.get("facts") or [])):
            return Rejection(raw, UNKNOWN_REFERENCE,
                             f"no fact {raw['fact_id']!r} to invalidate")
    if event_type == "resolve_story_thread":
        if raw["thread"] not in (state.get("threads") or {}):
            return Rejection(raw, UNKNOWN_REFERENCE,
                             f"no story thread {raw['thread']!r} to resolve")
    if spec["creates"]:
        key = raw[spec["creates"]]
        if key in known:
            return Rejection(raw, DUPLICATE_ENTITY,
                             f"{key!r} already exists; use set_* to change it")

    # ---- layer 5: semantics ----
    return _semantic(raw, state, canon)


def _bad_shape(value, kind: str, field: str) -> str | None:
    """Returns why `value` is the wrong shape for `kind`, or None."""
    if kind in (events.TEXT, events.KEY):
        if not isinstance(value, str) or not value.strip():
            return f"{field!r} must be a non-empty string"
        if len(value) > MAX_TEXT:
            return f"{field!r} is longer than {MAX_TEXT} characters"
        return None
    if kind == events.VALUE:
        # A scalar. Explicitly not a dict or a list: a nested payload is how a
        # value field becomes somewhere to hide a second protocol.
        if not isinstance(value, (str, int, float, bool)) and value is not None:
            return f"{field!r} must be a plain value, not a structure"
        if isinstance(value, str) and len(value) > MAX_TEXT:
            return f"{field!r} is longer than {MAX_TEXT} characters"
        return None
    if kind == events.LABELS:
        if not isinstance(value, list):
            return f"{field!r} must be a list"
        if len(value) > MAX_LABELS:
            return f"{field!r} has more than {MAX_LABELS} entries"
        for item in value:
            if not isinstance(item, str) or not item.strip():
                return f"{field!r} must contain only non-empty strings"
            if len(item) > MAX_TEXT:
                return f"{field!r} contains an over-long entry"
        return None
    return f"{field!r} has an unknown field kind"      # pragma: no cover


def _semantic(raw: dict, state: dict, canon: dict | None) -> Rejection | None:
    """Deterministic checks the application can actually make.

    Deliberately modest. ADR 010 is explicit that typed events do not make a
    model correct, and pretending arbitrary fiction can be validated would be
    worse than admitting it cannot: it would produce confident rejections of
    perfectly good story. So this refuses only what the application *knows* is
    wrong — a self-contradiction, or a collision with a rule the campaign wrote
    down.
    """
    event_type = raw["type"]

    # An entity cannot hold itself, and cannot be in itself.
    if event_type == "set_possession" and raw["item"] == raw["owner"]:
        return Rejection(raw, SELF_CONTRADICTION, "an item cannot possess itself")
    if event_type == "set_current_location" and raw["entity"] == raw["location"]:
        return Rejection(raw, SELF_CONTRADICTION, "an entity cannot be inside itself")
    if event_type in ("add_relationship", "end_relationship") and raw["source"] == raw["target"]:
        return Rejection(raw, SELF_CONTRADICTION,
                         "a relationship needs two different entities")

    # C01: campaign canon outranks narration. The rule is generic — a campaign
    # declares transitions it forbids, and any event proposing one is refused.
    # Nothing here knows what any of those transitions mean; the campaign
    # says which it forbids, in data.
    conflict = _canon_conflict(raw, state, canon)
    if conflict is not None:
        return Rejection(raw, CANON_CONFLICT, conflict)
    return None


def _canon_conflict(raw: dict, state: dict, canon: dict | None) -> str | None:
    """Whether campaign canon forbids what this event proposes.

    Canon is configuration, not code (J03). A campaign writes:

        {"forbidden_status_changes": [{"from": "dead", "to": "active"}]}

    and a narration that tries to bring a dead character back is refused —
    without this module, or any other, containing the word for what that is. A
    science-fiction campaign forbidding a different transition uses the same
    field and the same code path.
    """
    if not isinstance(canon, dict):
        return None
    if raw["type"] != "set_entity_status":
        return None
    forbidden = canon.get("forbidden_status_changes")
    if not isinstance(forbidden, list):
        return None
    current = (model.entity(state, raw["entity"]) or {}).get("status")
    for rule in forbidden:
        if not isinstance(rule, dict):
            continue
        if rule.get("from") == current and rule.get("to") == raw["status"]:
            return (
                f"campaign canon does not allow {raw['entity']!r} to go from "
                f"{current!r} to {raw['status']!r}"
            )
    return None